API Reference

This page documents the PDL members that likely to be used to run PDL programs from Python.

Program

PDL programs are represented by the Pydantic data structure defined in this file.

Classes:

Name	Description
`PdlLocationType`	Internal data structure to keep track of the source location information.
`LocalizedExpression`	Internal data structure for expressions with location information.
`Pattern`	Common fields for structured patterns.
`OrPattern`	Match any of the patterns.
`ArrayPattern`	Match an array.
`ObjectPattern`	Match an object.
`AnyPattern`	Match any value.
`PdlType`	Common fields for PDL types.
`OptionalPdlType`	Optional type.
`JsonSchemaTypePdlType`	Json Schema with a type field.
`EnumPdlType`	Json Schema with an `enum` field.
`ObjectPdlType`	Object type.
`Parser`	Common fields for all parsers (`parser` field).
`PdlParser`	Use a PDL program as a parser specification (experimental).
`RegexParser`	A regular expression parser.
`ContributeTarget`	Values allowed in the `contribute` field.
`ContributeValue`	Contribution of a specific value instead of the default one.
`ExpectationType`	Single expectation definition.
`PdlTiming`	Internal data structure to record timing information in the trace.
`PdlUsage`	Internal data structure to record token consumption usage information.
`Block`	Common fields for all PDL blocks.
`FunctionBlock`	Function declaration.
`CallBlock`	Calling a function.
`LitellmParameters`	Parameters passed to LiteLLM. More details at https://docs.litellm.ai/docs/completion/input.
`ModelBlock`	Common fields for the `model` blocks.
`LitellmModelBlock`	Call an LLM through the LiteLLM API.
`GraniteioProcessor`
`GraniteioModelBlock`	Call an LLM through the granite-io API.
`CodeBlock`	Execute a piece of code.
`ArgsBlock`	Execute a command line, which will spawn a subprocess with the given argument vector. Note: if you need a shell script execution, you must wrap your command line in /bin/sh or some shell of your choosing.
`GetBlock`	Get the value of a variable.
`DataBlock`	Arbitrary value, equivalent to JSON.
`TextBlock`	Create the concatenation of the stringify version of the result of each block of the list of blocks.
`LastOfBlock`	Return the value of the last block if the list of blocks.
`ArrayBlock`	Return the array of values computed by each block of the list of blocks.
`ObjectBlock`	Return the object where the value of each field is defined by a block. If the body of the object is an array, the resulting object is the union of the objects computed by each element of the array.
`MessageBlock`	Create a message.
`IfBlock`	Conditional control structure.
`MatchCase`	Case of a match.
`MatchBlock`	Match control structure.
`JoinConfig`	Configure how loop iterations should be combined.
`JoinText`	Join loop iterations as a string.
`JoinArray`	Join loop iterations as an array.
`JoinObject`	Join loop iterations as an object.
`JoinLastOf`	Join loop iterations as the value of the last iteration.
`JoinReduce`	Join loop iterations as the value of the last iteration.
`RepeatBlock`	Repeat the execution of a block sequentially.
`MapBlock`	Independent executions of a block.
`ReadBlock`	Read from a file or standard input.
`IncludeBlock`	Include a PDL file.
`ImportBlock`	Import a PDL file.
`AggregatorConfig`	Common fields for all aggregator configurations.
`FileAggregatorConfig`
`AggregatorBlock`	Create a new aggregator that can be use in the `contribute` field.
`ErrorBlock`	Block representing an error generated at runtime.
`EmptyBlock`	Block without an action. It can contain definitions.
`Program`	Prompt Declaration Language program (PDL)

Functions:

Name	Description
`get_default_model_parameters`	Model-specific defaults to apply
`get_sampling_defaults`	Model-specific defaults to apply if we are sampling.

Attributes:

Name	Type	Description
`OptionalStr`		Optional string.
`OptionalInt`		Optional integer.
`OptionalBool`		Optional Boolean.
`OptionalBoolOrStr`		Optional boolean or string.
`OptionalAny`		Optional value of any type.
`OptionalBlockType`		Optional block.
`ScopeType`		Type of the execution environment.
`ModelInput`		Type of the input of an LLM call.
`OptionalModelInput`		Optional value of type ModelInput.
`OptionalPdlLocationType`		Optional location type.
`ExpressionType`	`TypeAlias`	Expressions are represented Jinja as strings in between `${` and `}`.
`ExpressionStr`		Expression evaluating into a string.
`OptionalExpressionStr`		Optional expression evaluating into a string.
`ExpressionInt`		Expression evaluating into an int.
`OptionalExpressionInt`		Optional expression evaluating into an int.
`ExpressionBool`		Expression evaluating into a bool.
`OptionalExpressionBool`		Optional expression evaluating into a bool.
`PatternType`		Patterns allowed to match values in a `case` clause.
`BasePdlType`	`TypeAlias`	Base types.
`PdlTypeType`		Types.
`ParserType`		Different parsers.
`OptionalParserType`		Optional parser.
`RoleType`	`TypeAlias`	Role name.
`ContributeElement`		Type of the contribute field.
`ExpectationsType`		Type of expectations field
`OptionalPdlTiming`		Optional execution time information.
`OptionalPdlUsage`		Optional usage of statistics of an LLM call.
`JoinType`		Different ways to join loop iterations.
`AdvancedBlockType`	`TypeAlias`	Different types of structured blocks.
`BlockType`		All kinds of blocks.
`BlockOrBlocksType`	`TypeAlias`	Block or list of blocks.

`OptionalStr = TypeAliasType('OptionalStr', Optional[str])` `module-attribute`

Optional string.

`OptionalInt = TypeAliasType('OptionalInt', Optional[int])` `module-attribute`

Optional integer.

`OptionalBool = TypeAliasType('OptionalBool', Optional[bool])` `module-attribute`

Optional Boolean.

`OptionalBoolOrStr = TypeAliasType('OptionalBoolOrStr', Optional[Union[bool, str]])` `module-attribute`

Optional boolean or string.

`OptionalAny = TypeAliasType('OptionalAny', Optional[Any])` `module-attribute`

Optional value of any type.

`OptionalBlockType = TypeAliasType('OptionalBlockType', Optional['BlockType'])` `module-attribute`

Optional block.

`ScopeType = TypeAliasType('ScopeType', PdlDict[str, Any])` `module-attribute`

Type of the execution environment.

`ModelInput = TypeAliasType('ModelInput', Sequence[Mapping[str, Any]])` `module-attribute`

Type of the input of an LLM call.

`OptionalModelInput = TypeAliasType('OptionalModelInput', Optional[ModelInput])` `module-attribute`

Optional value of type ModelInput.

`OptionalPdlLocationType = TypeAliasType('OptionalPdlLocationType', Optional[PdlLocationType])` `module-attribute`

Optional location type.

`ExpressionType: TypeAlias = LocalizedExpression[ExpressionTypeT] | ExpressionTypeT | str` `module-attribute`

Expressions are represented Jinja as strings in between ${ and }.

`ExpressionStr = TypeAliasType('ExpressionStr', ExpressionType[str])` `module-attribute`

Expression evaluating into a string.

`OptionalExpressionStr = TypeAliasType('OptionalExpressionStr', Optional[ExpressionStr])` `module-attribute`

Optional expression evaluating into a string.

`ExpressionInt = TypeAliasType('ExpressionInt', ExpressionType[int])` `module-attribute`

Expression evaluating into an int.

`OptionalExpressionInt = TypeAliasType('OptionalExpressionInt', Optional[ExpressionInt])` `module-attribute`

Optional expression evaluating into an int.

`ExpressionBool = TypeAliasType('ExpressionBool', ExpressionType[bool])` `module-attribute`

Expression evaluating into a bool.

`OptionalExpressionBool = TypeAliasType('OptionalExpressionBool', Optional[ExpressionBool])` `module-attribute`

Optional expression evaluating into a bool.

`PatternType = TypeAliasType('PatternType', Union[None, bool, int, float, str, OrPattern, ArrayPattern, ObjectPattern, AnyPattern])` `module-attribute`

Patterns allowed to match values in a case clause.

`BasePdlType: TypeAlias = Literal['null', 'boolean', 'string', 'number', 'integer', 'array', 'object', 'bool', 'str', 'float', 'int', 'list', 'obj']` `module-attribute`

Base types.

`PdlTypeType = TypeAliasType('PdlTypeType', Annotated["Union[None, BasePdlType, EnumPdlType, list['PdlTypeType'], OptionalPdlType, JsonSchemaTypePdlType, ObjectPdlType, dict[str, 'PdlTypeType']]", Field(union_mode='left_to_right')])` `module-attribute`

Types.

`ParserType = TypeAliasType('ParserType', Union[Literal['json', 'jsonl', 'yaml'], PdlParser, RegexParser])` `module-attribute`

Different parsers.

`OptionalParserType = TypeAliasType('OptionalParserType', Optional[ParserType])` `module-attribute`

Optional parser.

`RoleType: TypeAlias = OptionalStr` `module-attribute`

Role name.

`ContributeElement = TypeAliasType('ContributeElement', Union[ContributeTarget, str, dict[str, ContributeValue]])` `module-attribute`

Type of the contribute field.

`ExpectationsType = TypeAliasType('ExpectationsType', Sequence[ExpectationType])` `module-attribute`

Type of expectations field

`OptionalPdlTiming = TypeAliasType('OptionalPdlTiming', Optional[PdlTiming])` `module-attribute`

Optional execution time information.

`OptionalPdlUsage = TypeAliasType('OptionalPdlUsage', Optional[PdlUsage])` `module-attribute`

Optional usage of statistics of an LLM call.

`JoinType = TypeAliasType('JoinType', Union[JoinText, JoinArray, JoinObject, JoinLastOf, JoinReduce])` `module-attribute`

Different ways to join loop iterations.

`AdvancedBlockType: TypeAlias = FunctionBlock | CallBlock | LitellmModelBlock | GraniteioModelBlock | CodeBlock | ArgsBlock | GetBlock | DataBlock | IfBlock | MatchBlock | RepeatBlock | MapBlock | TextBlock | LastOfBlock | ArrayBlock | ObjectBlock | MessageBlock | ReadBlock | IncludeBlock | ImportBlock | AggregatorBlock | ErrorBlock | EmptyBlock` `module-attribute`

Different types of structured blocks.

`BlockType = TypeAliasType('BlockType', Union[None, bool, int, float, str, AdvancedBlockType])` `module-attribute`

All kinds of blocks.

`BlockOrBlocksType: TypeAlias = BlockType | list[BlockType]` `module-attribute`

Block or list of blocks.

`PdlLocationType`

Bases: BaseModel

Internal data structure to keep track of the source location information.

Source code in src/pdl/pdl_ast.py

class PdlLocationType(BaseModel):
    """Internal data structure to keep track of the source location information."""

    model_config = ConfigDict(extra="forbid")
    path: list[str]
    file: str
    table: dict[str, int]

`LocalizedExpression`

Bases: BaseModel, Generic[LocalizedExpressionT]

Internal data structure for expressions with location information.

Source code in src/pdl/pdl_ast.py

class LocalizedExpression(BaseModel, Generic[LocalizedExpressionT]):
    """Internal data structure for expressions with location information."""

    model_config = ConfigDict(
        extra="forbid",
        use_attribute_docstrings=True,
        arbitrary_types_allowed=True,
        model_title_generator=(lambda _: "LocalizedExpression"),
    )
    pdl__expr: Any
    pdl__result: Optional[LocalizedExpressionT] = None
    pdl__location: OptionalPdlLocationType = None

`Pattern`

Bases: BaseModel

Common fields for structured patterns.

Attributes:

Name	Type	Description
`def_`	`OptionalStr`	Name of the variable used to store the value matched by the pattern.

Source code in src/pdl/pdl_ast.py

class Pattern(BaseModel):
    """Common fields for structured patterns."""

    model_config = ConfigDict(extra="forbid")
    def_: OptionalStr = Field(default=None, alias="def")
    """Name of the variable used to store the value matched by the pattern.
    """

`def_: OptionalStr = Field(default=None, alias='def')` `class-attribute` `instance-attribute`

Name of the variable used to store the value matched by the pattern.

`OrPattern`

Bases: Pattern

Match any of the patterns.

Attributes:

Name	Type	Description
`anyOf`	`list[PatternType]`	List of possible patterns.

Source code in src/pdl/pdl_ast.py

class OrPattern(Pattern):
    """Match any of the patterns."""

    anyOf: list["PatternType"]
    """List of possible patterns."""

`anyOf: list[PatternType]` `instance-attribute`

List of possible patterns.

`ArrayPattern`

Bases: Pattern

Match an array.

Attributes:

Name	Type	Description
`array`	`list[PatternType]`	Shape of the array to match.

Source code in src/pdl/pdl_ast.py

class ArrayPattern(Pattern):
    """Match an array."""

    array: list["PatternType"]
    """Shape of the array to match."""

`array: list[PatternType]` `instance-attribute`

Shape of the array to match.

`ObjectPattern`

Bases: Pattern

Match an object.

Attributes:

Name	Type	Description
`object`	`dict[str, PatternType]`	Shape of the object to match.

Source code in src/pdl/pdl_ast.py

class ObjectPattern(Pattern):
    """Match an object."""

    object: dict[str, "PatternType"]
    """Shape of the object to match."""

`object: dict[str, PatternType]` `instance-attribute`

Shape of the object to match.

`AnyPattern`

Bases: Pattern

Match any value.

Source code in src/pdl/pdl_ast.py

class AnyPattern(Pattern):
    """Match any value."""

    any: Literal[None]

`PdlType`

Bases: BaseModel

Common fields for PDL types.

Source code in src/pdl/pdl_ast.py

class PdlType(BaseModel):
    """Common fields for PDL types."""

    model_config = ConfigDict(extra="forbid")

`OptionalPdlType`

Bases: PdlType

Optional type.

Attributes:

Name	Type	Description
`optional`	`PdlTypeType`

Source code in src/pdl/pdl_ast.py

class OptionalPdlType(PdlType):
    """Optional type."""

    optional: "PdlTypeType"
    """"""

`optional: PdlTypeType` `instance-attribute`

`JsonSchemaTypePdlType`

Bases: PdlType

Json Schema with a type field.

Attributes:

Name	Type	Description
`type`	`str \| list[str]`	Data type that a schema should expect.

Source code in src/pdl/pdl_ast.py

class JsonSchemaTypePdlType(PdlType):
    """Json Schema with a type field."""

    model_config = ConfigDict(extra="allow")

    type: str | list[str]
    """Data type that a schema should expect."""

`type: str | list[str]` `instance-attribute`

Data type that a schema should expect.

`EnumPdlType`

Bases: PdlType

Json Schema with an enum field.

Attributes:

Name	Type	Description
`enum`	`list[Any]`	List of allowed values in the type.

Source code in src/pdl/pdl_ast.py

class EnumPdlType(PdlType):
    """Json Schema with an `enum` field."""

    model_config = ConfigDict(extra="allow")

    enum: list[Any]
    """List of allowed values in the type."""

`enum: list[Any]` `instance-attribute`

List of allowed values in the type.

`ObjectPdlType`

Bases: PdlType

Object type.

Attributes:

Name	Type	Description
`object`	`dict[str, PdlTypeType]`	Fields of the objects with their types.

Source code in src/pdl/pdl_ast.py

class ObjectPdlType(PdlType):
    """Object type."""

    object: dict[str, "PdlTypeType"]
    """Fields of the objects with their types."""

`object: dict[str, PdlTypeType]` `instance-attribute`

Fields of the objects with their types.

`Parser`

Bases: BaseModel

Common fields for all parsers (parser field).

Attributes:

Name	Type	Description
`description`	`OptionalStr`	Documentation associated to the parser.
`spec`	`PdlTypeType`	Expected type of the parsed value.

Source code in src/pdl/pdl_ast.py

class Parser(BaseModel):
    """Common fields for all parsers (`parser` field)."""

    model_config = ConfigDict(extra="forbid")
    description: OptionalStr = None
    """Documentation associated to the parser.
    """
    spec: PdlTypeType = None
    """Expected type of the parsed value.
    """

`description: OptionalStr = None` `class-attribute` `instance-attribute`

Documentation associated to the parser.

`spec: PdlTypeType = None` `class-attribute` `instance-attribute`

Expected type of the parsed value.

`PdlParser`

Bases: Parser

Use a PDL program as a parser specification (experimental).

Attributes:

Name	Type	Description
`pdl`	`BlockType`	PDL program describing the shape of the expected value.

Source code in src/pdl/pdl_ast.py

class PdlParser(Parser):
    """Use a PDL program as a parser specification (experimental)."""

    pdl: "BlockType"
    """PDL program describing the shape of the expected value."""

`pdl: BlockType` `instance-attribute`

PDL program describing the shape of the expected value.

`RegexParser`

Bases: Parser

A regular expression parser.

Attributes:

Name	Type	Description
`regex`	`str`	Regular expression to parse the value.
`mode`	`Annotated[Literal['search', 'match', 'fullmatch', 'split', 'findall'], BeforeValidator(_ensure_lower)]`	Function used to parse to value (https://docs.python.org/3/library/re.html).

Source code in src/pdl/pdl_ast.py

class RegexParser(Parser):
    """A regular expression parser."""

    regex: str
    """Regular expression to parse the value."""
    mode: Annotated[
        Literal["search", "match", "fullmatch", "split", "findall"],
        BeforeValidator(_ensure_lower),
    ] = "fullmatch"
    """Function used to parse to value (https://docs.python.org/3/library/re.html)."""

`regex: str` `instance-attribute`

Regular expression to parse the value.

`mode: Annotated[Literal['search', 'match', 'fullmatch', 'split', 'findall'], BeforeValidator(_ensure_lower)] = 'fullmatch'` `class-attribute` `instance-attribute`

Function used to parse to value (https://docs.python.org/3/library/re.html).

`ContributeTarget`

Bases: StrEnum

Values allowed in the contribute field.

Source code in src/pdl/pdl_ast.py

class ContributeTarget(StrEnum):
    """Values allowed in the `contribute` field."""

    RESULT = "result"
    CONTEXT = "context"
    STDOUT = "stdout"
    STDERR = "stderr"

`ContributeValue`

Bases: BaseModel

Contribution of a specific value instead of the default one.

Attributes:

Name	Type	Description
`value`	`ExpressionType[Any]`	Value to contribute.

Source code in src/pdl/pdl_ast.py

class ContributeValue(BaseModel):
    """Contribution of a specific value instead of the default one."""

    model_config = ConfigDict(extra="forbid")

    value: ExpressionType[Any]
    """Value to contribute."""

`value: ExpressionType[Any]` `instance-attribute`

Value to contribute.

`ExpectationType`

Bases: BaseModel

Single expectation definition.

Attributes:

Name	Type	Description
`expect`	`ExpressionType`	English description of the expectation
`feedback`	`Optional[ExpressionType[FunctionBlock]]`	Feedback function for the expectation

Source code in src/pdl/pdl_ast.py

class ExpectationType(BaseModel):
    """Single expectation definition."""

    model_config = ConfigDict(extra="forbid")

    expect: ExpressionType
    """English description of the expectation"""

    feedback: Optional[ExpressionType["FunctionBlock"]] = None
    """Feedback function for the expectation"""

`expect: ExpressionType` `instance-attribute`

English description of the expectation

`feedback: Optional[ExpressionType[FunctionBlock]] = None` `class-attribute` `instance-attribute`

Feedback function for the expectation

`PdlTiming`

Bases: BaseModel

Internal data structure to record timing information in the trace.

Attributes:

Name	Type	Description
`start_nanos`	`OptionalInt`	Time at which block execution began.
`end_nanos`	`OptionalInt`	Time at which block execution ended.
`first_use_nanos`	`OptionalInt`	Time at which the value of the block was needed for the first time.
`timezone`	`OptionalStr`	Timezone of start_nanos and end_nanos.

Source code in src/pdl/pdl_ast.py

class PdlTiming(BaseModel):
    """Internal data structure to record timing information in the trace."""

    model_config = ConfigDict(extra="forbid")
    start_nanos: OptionalInt = 0
    """Time at which block execution began.
    """
    end_nanos: OptionalInt = 0
    """Time at which block execution ended.
    """
    first_use_nanos: OptionalInt = 0
    """Time at which the value of the block was needed for the first time.
    """
    timezone: OptionalStr = ""
    """Timezone of start_nanos and end_nanos.
    """

`start_nanos: OptionalInt = 0` `class-attribute` `instance-attribute`

Time at which block execution began.

`end_nanos: OptionalInt = 0` `class-attribute` `instance-attribute`

Time at which block execution ended.

`first_use_nanos: OptionalInt = 0` `class-attribute` `instance-attribute`

Time at which the value of the block was needed for the first time.

`timezone: OptionalStr = ''` `class-attribute` `instance-attribute`

Timezone of start_nanos and end_nanos.

`PdlUsage`

Bases: BaseModel

Internal data structure to record token consumption usage information.

Attributes:

Name	Type	Description
`completion_tokens`	`OptionalInt`	Completion tokens consumed
`prompt_tokens`	`OptionalInt`	Prompt tokens consumed

Source code in src/pdl/pdl_ast.py

class PdlUsage(BaseModel):
    """Internal data structure to record token consumption usage information."""

    completion_tokens: OptionalInt = 0
    """Completion tokens consumed
    """
    prompt_tokens: OptionalInt = 0
    """Prompt tokens consumed
    """

`completion_tokens: OptionalInt = 0` `class-attribute` `instance-attribute`

Completion tokens consumed

`prompt_tokens: OptionalInt = 0` `class-attribute` `instance-attribute`

Prompt tokens consumed

`Block`

Bases: BaseModel

Common fields for all PDL blocks.

Attributes:

Name	Type	Description
`description`	`OptionalStr`	Documentation associated to the block.
`spec`	`PdlTypeType`	Type specification of the result of the block.
`defs`	`dict[str, BlockType]`	Set of definitions executed before the execution of the block.
`def_`	`OptionalStr`	Name of the variable used to store the result of the execution of the block.
`contribute`	`Sequence[ContributeElement]`	Indicate if the block contributes to the result and background context.
`parser`	`Annotated[OptionalParserType, BeforeValidator(_ensure_lower)]`	Parser to use to construct a value out of a string result.
`fallback`	`OptionalBlockType`	Block to execute in case of error.
`retry`	`OptionalInt`	The maximum number of times to retry when an error occurs within a block.
`trace_error_on_retry`	`OptionalBoolOrStr`	Whether to add the errors while retrying to the trace. Set this to true to use retry feature for multiple LLM trials.
`expectations`	`ExpectationsType`	Specify any expectations that the result of the block must satisfy.
`role`	`RoleType`	Role associated to the block and sub-blocks.
`pdl__context`	`OptionalModelInput`	Current context.
`pdl__id`	`OptionalStr`	Unique identifier for this block.
`pdl__result`	`OptionalAny`	Result of the execution of the block.
`pdl__timing`	`OptionalPdlTiming`	Execution timing information.

Source code in src/pdl/pdl_ast.py

class Block(BaseModel):
    """Common fields for all PDL blocks."""

    model_config = ConfigDict(
        extra="forbid",
        use_attribute_docstrings=True,
        arbitrary_types_allowed=True,
        validate_by_name=True,
    )

    description: OptionalStr = None
    """Documentation associated to the block.
    """
    spec: PdlTypeType = None
    """Type specification of the result of the block.
    """
    defs: dict[str, "BlockType"] = {}
    """Set of definitions executed before the execution of the block.
    """
    def_: OptionalStr = Field(default=None, alias="def")
    """Name of the variable used to store the result of the execution of the block.
    """
    contribute: Sequence[ContributeElement] = [
        ContributeTarget.RESULT,
        ContributeTarget.CONTEXT,
    ]
    """Indicate if the block contributes to the result and background context.
    """
    parser: Annotated[OptionalParserType, BeforeValidator(_ensure_lower)] = None
    """Parser to use to construct a value out of a string result."""
    fallback: OptionalBlockType = None
    """Block to execute in case of error.
    """
    retry: OptionalInt = None
    """The maximum number of times to retry when an error occurs within a block.
    """
    trace_error_on_retry: OptionalBoolOrStr = None
    """Whether to add the errors while retrying to the trace. Set this to true to use retry feature for multiple LLM trials.
    """

    expectations: ExpectationsType = []
    """Specify any expectations that the result of the block must satisfy.
    """

    role: RoleType = None
    """Role associated to the block and sub-blocks.
    Typical roles are `system`, `user`, and `assistant`,
    but there may be other roles such as `available_tools`.
    """
    # Fields for internal use
    pdl__context: OptionalModelInput = []
    """Current context."""
    pdl__id: OptionalStr = ""
    """Unique identifier for this block."""
    pdl__result: OptionalAny = None
    """Result of the execution of the block."""
    pdl__location: OptionalPdlLocationType = None
    pdl__timing: OptionalPdlTiming = None
    """Execution timing information."""

`description: OptionalStr = None` `class-attribute` `instance-attribute`

Documentation associated to the block.

`spec: PdlTypeType = None` `class-attribute` `instance-attribute`

Type specification of the result of the block.

`defs: dict[str, BlockType] = {}` `class-attribute` `instance-attribute`

Set of definitions executed before the execution of the block.

`def_: OptionalStr = Field(default=None, alias='def')` `class-attribute` `instance-attribute`

Name of the variable used to store the result of the execution of the block.

`contribute: Sequence[ContributeElement] = [ContributeTarget.RESULT, ContributeTarget.CONTEXT]` `class-attribute` `instance-attribute`

Indicate if the block contributes to the result and background context.

`parser: Annotated[OptionalParserType, BeforeValidator(_ensure_lower)] = None` `class-attribute` `instance-attribute`

Parser to use to construct a value out of a string result.

`fallback: OptionalBlockType = None` `class-attribute` `instance-attribute`

Block to execute in case of error.

`retry: OptionalInt = None` `class-attribute` `instance-attribute`

The maximum number of times to retry when an error occurs within a block.

`trace_error_on_retry: OptionalBoolOrStr = None` `class-attribute` `instance-attribute`

Whether to add the errors while retrying to the trace. Set this to true to use retry feature for multiple LLM trials.

`expectations: ExpectationsType = []` `class-attribute` `instance-attribute`

Specify any expectations that the result of the block must satisfy.

`role: RoleType = None` `class-attribute` `instance-attribute`

Role associated to the block and sub-blocks. Typical roles are system, user, and assistant, but there may be other roles such as available_tools.

`pdl__context: OptionalModelInput = []` `class-attribute` `instance-attribute`

Current context.

`pdl__id: OptionalStr = ''` `class-attribute` `instance-attribute`

Unique identifier for this block.

`pdl__result: OptionalAny = None` `class-attribute` `instance-attribute`

Result of the execution of the block.

`pdl__timing: OptionalPdlTiming = None` `class-attribute` `instance-attribute`

Execution timing information.

`FunctionBlock`

Bases: LeafBlock

Function declaration.

Attributes:

Name	Type	Description
`function`	`Optional[dict[str, PdlTypeType]]`	Functions parameters with their types.
`return_`	`BlockType`	Body of the function.
`signature`	`Optional[Json]`	Function signature computed from the function definition.

Source code in src/pdl/pdl_ast.py

class FunctionBlock(LeafBlock):
    """Function declaration."""

    kind: Literal[BlockKind.FUNCTION] = BlockKind.FUNCTION
    function: Optional[dict[str, PdlTypeType]]
    """Functions parameters with their types.
    """
    return_: "BlockType" = Field(..., alias="return")
    """Body of the function.
    """
    signature: Optional[Json] = None
    """Function signature computed from the function definition.
    """

`function: Optional[dict[str, PdlTypeType]]` `instance-attribute`

Functions parameters with their types.

`return_: BlockType = Field(..., alias='return')` `class-attribute` `instance-attribute`

Body of the function.

`signature: Optional[Json] = None` `class-attribute` `instance-attribute`

Function signature computed from the function definition.

`CallBlock`

Bases: LeafBlock

Calling a function.

Attributes:

Name	Type	Description
`call`	`ExpressionType[FunctionBlock]`	Function to call.
`args`	`ExpressionType`	Arguments of the function with their values.

Source code in src/pdl/pdl_ast.py

class CallBlock(LeafBlock):
    """Calling a function."""

    kind: Literal[BlockKind.CALL] = BlockKind.CALL
    call: ExpressionType[FunctionBlock]
    """Function to call.
    """
    args: ExpressionType = {}
    """Arguments of the function with their values.
    """
    # Field for internal use
    pdl__trace: OptionalBlockType = None

`call: ExpressionType[FunctionBlock]` `instance-attribute`

Function to call.

`args: ExpressionType = {}` `class-attribute` `instance-attribute`

Arguments of the function with their values.

`LitellmParameters`

Bases: BaseModel

Parameters passed to LiteLLM. More details at https://docs.litellm.ai/docs/completion/input.

Note that not all models and platforms accept all parameters.

Attributes:

Name	Type	Description
`timeout`	`Optional[Union[float, str]] \| str`	Timeout in seconds for completion requests (Defaults to 600 seconds).
`temperature`	`Optional[float] \| str`	The temperature parameter for controlling the randomness of the output (default is 1.0).
`top_p`	`Optional[float] \| str`	The top-p parameter for nucleus sampling (default is 1.0).
`n`	`Optional[int] \| str`	The number of completions to generate (default is 1).
`stop`	`Optional[str \| list[str]] \| str`	Up to 4 sequences where the LLM API will stop generating further tokens.
`max_tokens`	`Optional[int] \| str`	The maximum number of tokens in the generated completion (default is infinity).
`presence_penalty`	`Optional[float] \| str`	It is used to penalize new tokens based on their existence in the text so far.
`frequency_penalty`	`Optional[float] \| str`	It is used to penalize new tokens based on their frequency in the text so far.
`logit_bias`	`Optional[dict] \| str`	Used to modify the probability of specific tokens appearing in the completion.
`user`	`Optional[str] \| str`	A unique identifier representing your end-user. This can help the LLM provider to monitor and detect abuse.
`logprobs`	`Optional[bool] \| str`	Whether to return log probabilities of the output tokens or not. If true, returns the log probabilities of each output token returned in the content of message
`top_logprobs`	`Optional[int] \| str`	top_logprobs (int, optional): An integer between 0 and 5 specifying the number of most likely tokens to return at each token position, each with an associated log probability. logprobs must be set to true if this parameter is used.
`extra_headers`	`Optional[dict] \| str`	Additional headers to include in the request.
`functions`	`Optional[list] \| str`	A list of functions to apply to the conversation messages (default is an empty list)
`function_call`	`Optional[str] \| str`	The name of the function to call within the conversation (default is an empty string)
`base_url`	`Optional[str] \| str`	Base URL for the API (default is None).
`api_version`	`Optional[str] \| str`	API version (default is None).
`api_key`	`Optional[str] \| str`	API key (default is None).
`model_list`	`Optional[list] \| str`	List of api base, version, keys.
`mock_response`	`Optional[str] \| str`	If provided, return a mock completion response for testing or debugging purposes (default is None).
`custom_llm_provider`	`Optional[str] \| str`	Used for Non-OpenAI LLMs, Example usage for bedrock, set model="amazon.titan-tg1-large" and custom_llm_provider="bedrock"
`max_retries`	`Optional[int] \| str`	The number of retries to attempt (default is 0).

Source code in src/pdl/pdl_ast.py

class LitellmParameters(BaseModel):
    """Parameters passed to LiteLLM. More details at [https://docs.litellm.ai/docs/completion/input](https://docs.litellm.ai/docs/completion/input).

    Note that not all models and platforms accept all parameters.
    """

    model_config = ConfigDict(extra="allow", protected_namespaces=())
    timeout: Optional[Union[float, str]] | str = None
    """Timeout in seconds for completion requests (Defaults to 600 seconds).
    """
    temperature: Optional[float] | str = None
    """The temperature parameter for controlling the randomness of the output (default is 1.0).
    """
    top_p: Optional[float] | str = None
    """The top-p parameter for nucleus sampling (default is 1.0).
    """
    n: Optional[int] | str = None
    """The number of completions to generate (default is 1).
    """
    # stream: Optional[bool] = None
    # """If True, return a streaming response (default is False).
    # """
    # stream_options: Optional[dict] = None
    # """A dictionary containing options for the streaming response. Only set this when you set stream: true.
    # """
    stop: Optional[str | list[str]] | str = None
    """Up to 4 sequences where the LLM API will stop generating further tokens.
    """
    max_tokens: Optional[int] | str = None
    """The maximum number of tokens in the generated completion (default is infinity).
    """
    presence_penalty: Optional[float] | str = None
    """It is used to penalize new tokens based on their existence in the text so far.
    """
    frequency_penalty: Optional[float] | str = None
    """It is used to penalize new tokens based on their frequency in the text so far.
    """
    logit_bias: Optional[dict] | str = None
    """Used to modify the probability of specific tokens appearing in the completion.
    """
    user: Optional[str] | str = None
    """A unique identifier representing your end-user. This can help the LLM provider to monitor and detect abuse.
    """
    # openai v1.0+ new params
    response_format: Optional[dict] | str = None
    seed: Optional[int] | str = None
    tools: Optional[list] | str = None
    tool_choice: Optional[Union[str, dict]] | str = None
    logprobs: Optional[bool] | str = None
    """Whether to return log probabilities of the output tokens or not. If true, returns the log probabilities of each output token returned in the content of message
    """
    top_logprobs: Optional[int] | str = None
    """top_logprobs (int, optional): An integer between 0 and 5 specifying the number of most likely tokens to return at each token position, each with an associated log probability. logprobs must be set to true if this parameter is used.
    """
    parallel_tool_calls: Optional[bool] | str = None
    # deployment_id = None
    extra_headers: Optional[dict] | str = None
    """Additional headers to include in the request.
    """
    # soon to be deprecated params by OpenAI
    functions: Optional[list] | str = None
    """A list of functions to apply to the conversation messages (default is an empty list)
    """
    function_call: Optional[str] | str = None
    """The name of the function to call within the conversation (default is an empty string)
    """
    # set api_base, api_version, api_key
    base_url: Optional[str] | str = environ.get("OPENAI_BASE_URL")
    """Base URL for the API (default is None).
    """
    api_version: Optional[str] | str = None
    """API version (default is None).
    """
    api_key: Optional[str] | str = None
    """API key (default is None).
    """
    model_list: Optional[list] | str = None  # pass in a list of api_base,keys, etc.
    """List of api base, version, keys.
    """
    # Optional liteLLM function params
    mock_response: Optional[str] | str = None
    """If provided, return a mock completion response for testing or debugging purposes (default is None).
    """
    custom_llm_provider: Optional[str] | str = None
    """Used for Non-OpenAI LLMs, Example usage for bedrock, set model="amazon.titan-tg1-large" and custom_llm_provider="bedrock"
    """
    max_retries: Optional[int] | str = None
    """The number of retries to attempt (default is 0).
    """

`timeout: Optional[Union[float, str]] | str = None` `class-attribute` `instance-attribute`

Timeout in seconds for completion requests (Defaults to 600 seconds).

`temperature: Optional[float] | str = None` `class-attribute` `instance-attribute`

The temperature parameter for controlling the randomness of the output (default is 1.0).

`top_p: Optional[float] | str = None` `class-attribute` `instance-attribute`

The top-p parameter for nucleus sampling (default is 1.0).

`n: Optional[int] | str = None` `class-attribute` `instance-attribute`

The number of completions to generate (default is 1).

`stop: Optional[str | list[str]] | str = None` `class-attribute` `instance-attribute`

Up to 4 sequences where the LLM API will stop generating further tokens.

`max_tokens: Optional[int] | str = None` `class-attribute` `instance-attribute`

The maximum number of tokens in the generated completion (default is infinity).

`presence_penalty: Optional[float] | str = None` `class-attribute` `instance-attribute`

It is used to penalize new tokens based on their existence in the text so far.

`frequency_penalty: Optional[float] | str = None` `class-attribute` `instance-attribute`

It is used to penalize new tokens based on their frequency in the text so far.

`logit_bias: Optional[dict] | str = None` `class-attribute` `instance-attribute`

Used to modify the probability of specific tokens appearing in the completion.

`user: Optional[str] | str = None` `class-attribute` `instance-attribute`

A unique identifier representing your end-user. This can help the LLM provider to monitor and detect abuse.

`logprobs: Optional[bool] | str = None` `class-attribute` `instance-attribute`

Whether to return log probabilities of the output tokens or not. If true, returns the log probabilities of each output token returned in the content of message

`top_logprobs: Optional[int] | str = None` `class-attribute` `instance-attribute`

top_logprobs (int, optional): An integer between 0 and 5 specifying the number of most likely tokens to return at each token position, each with an associated log probability. logprobs must be set to true if this parameter is used.

`extra_headers: Optional[dict] | str = None` `class-attribute` `instance-attribute`

Additional headers to include in the request.

`functions: Optional[list] | str = None` `class-attribute` `instance-attribute`

A list of functions to apply to the conversation messages (default is an empty list)

`function_call: Optional[str] | str = None` `class-attribute` `instance-attribute`

The name of the function to call within the conversation (default is an empty string)

`base_url: Optional[str] | str = environ.get('OPENAI_BASE_URL')` `class-attribute` `instance-attribute`

Base URL for the API (default is None).

`api_version: Optional[str] | str = None` `class-attribute` `instance-attribute`

API version (default is None).

`api_key: Optional[str] | str = None` `class-attribute` `instance-attribute`

API key (default is None).

`model_list: Optional[list] | str = None` `class-attribute` `instance-attribute`

List of api base, version, keys.

`mock_response: Optional[str] | str = None` `class-attribute` `instance-attribute`

If provided, return a mock completion response for testing or debugging purposes (default is None).

`custom_llm_provider: Optional[str] | str = None` `class-attribute` `instance-attribute`

Used for Non-OpenAI LLMs, Example usage for bedrock, set model="amazon.titan-tg1-large" and custom_llm_provider="bedrock"

`max_retries: Optional[int] | str = None` `class-attribute` `instance-attribute`

The number of retries to attempt (default is 0).

`ModelBlock`

Bases: LeafBlock

Common fields for the model blocks.

Attributes:

Name	Type	Description
`input`	`BlockType`	Messages to send to the model.
`modelResponse`	`OptionalStr`	Variable where to store the raw response of the model.
`pdl__usage`	`OptionalPdlUsage`	Tokens consumed during model call

Source code in src/pdl/pdl_ast.py

class ModelBlock(LeafBlock):
    """Common fields for the `model` blocks."""

    kind: Literal[BlockKind.MODEL] = BlockKind.MODEL
    input: "BlockType" = "${ pdl_context }"
    """Messages to send to the model.
    """
    modelResponse: OptionalStr = None
    """Variable where to store the raw response of the model.
    """
    # Field for internal use
    pdl__usage: OptionalPdlUsage = None
    """Tokens consumed during model call
    """
    pdl__model_input: OptionalModelInput = None

`input: BlockType = '${ pdl_context }'` `class-attribute` `instance-attribute`

Messages to send to the model.

`modelResponse: OptionalStr = None` `class-attribute` `instance-attribute`

Variable where to store the raw response of the model.

`pdl__usage: OptionalPdlUsage = None` `class-attribute` `instance-attribute`

Tokens consumed during model call

`LitellmModelBlock`

Bases: ModelBlock

Call an LLM through the LiteLLM API.

Example:

model: ollama/granite-code:8b
parameters:
  stop: ['!']

Attributes:

Name	Type	Description
`platform`	`Literal[LITELLM]`	Optional field to ensure that the block is using LiteLLM.
`model`	`ExpressionStr`	Name of the model following the LiteLLM convention.
`parameters`	`Optional[LitellmParameters \| ExpressionType[dict]]`	Parameters to send to the model.
`structuredDecoding`	`OptionalBool`	Perform structured decoding if possible (i.e., `parser` and `spec` are provided and the inference platform supports it).

Source code in src/pdl/pdl_ast.py

class LitellmModelBlock(ModelBlock):
    """
    Call an LLM through [the LiteLLM API](https://docs.litellm.ai/).

    Example:
    ```PDL
    model: ollama/granite-code:8b
    parameters:
      stop: ['!']
    ```
    """

    platform: Literal[ModelPlatform.LITELLM] = ModelPlatform.LITELLM
    """Optional field to ensure that the block is using LiteLLM.
    """
    model: ExpressionStr
    """Name of the model following the LiteLLM convention.
    """
    parameters: Optional[LitellmParameters | ExpressionType[dict]] = None
    """Parameters to send to the model.
    """
    structuredDecoding: OptionalBool = True
    """Perform structured decoding if possible (i.e., `parser` and `spec` are provided and the inference platform supports it)."""

`platform: Literal[ModelPlatform.LITELLM] = ModelPlatform.LITELLM` `class-attribute` `instance-attribute`

Optional field to ensure that the block is using LiteLLM.

`model: ExpressionStr` `instance-attribute`

Name of the model following the LiteLLM convention.

`parameters: Optional[LitellmParameters | ExpressionType[dict]] = None` `class-attribute` `instance-attribute`

Parameters to send to the model.

`structuredDecoding: OptionalBool = True` `class-attribute` `instance-attribute`

Perform structured decoding if possible (i.e., parser and spec are provided and the inference platform supports it).

`GraniteioProcessor`

Bases: BaseModel

Attributes:

Name	Type	Description
`type`	`OptionalExpressionStr`	Type of IO processor.
`model`	`OptionalExpressionStr`	Model name used by the backend.
`backend`	`ExpressionType[str \| dict[str, Any] \| object]`	Backend object or name and configuration.

Source code in src/pdl/pdl_ast.py

class GraniteioProcessor(BaseModel):
    type: OptionalExpressionStr = None
    """Type of IO processor.
    """
    model: OptionalExpressionStr = None
    """Model name used by the backend.
    """
    backend: ExpressionType[str | dict[str, Any] | object]
    """Backend object or name and configuration.
    """

`type: OptionalExpressionStr = None` `class-attribute` `instance-attribute`

Type of IO processor.

`model: OptionalExpressionStr = None` `class-attribute` `instance-attribute`

Model name used by the backend.

`backend: ExpressionType[str | dict[str, Any] | object]` `instance-attribute`

Backend object or name and configuration.

`GraniteioModelBlock`

Bases: ModelBlock

Call an LLM through the granite-io API.

Attributes:

Name	Type	Description
`platform`	`Literal[GRANITEIO]`	Optional field to ensure that the block is using granite-io.
`processor`	`GraniteioProcessor \| ExpressionType[object]`	IO Processor configuration or object.
`parameters`	`Optional[ExpressionType[dict[str, Any]]]`	Parameters sent to the model.

Source code in src/pdl/pdl_ast.py

class GraniteioModelBlock(ModelBlock):
    """Call an LLM through the granite-io API."""

    platform: Literal[ModelPlatform.GRANITEIO] = ModelPlatform.GRANITEIO
    """Optional field to ensure that the block is using granite-io.
    """
    processor: GraniteioProcessor | ExpressionType[object]
    """IO Processor configuration or object.
    """
    parameters: Optional[ExpressionType[dict[str, Any]]] = None
    """Parameters sent to the model.
    """

`platform: Literal[ModelPlatform.GRANITEIO] = ModelPlatform.GRANITEIO` `class-attribute` `instance-attribute`

Optional field to ensure that the block is using granite-io.

`processor: GraniteioProcessor | ExpressionType[object]` `instance-attribute`

IO Processor configuration or object.

`parameters: Optional[ExpressionType[dict[str, Any]]] = None` `class-attribute` `instance-attribute`

Parameters sent to the model.

`CodeBlock`

Bases: BaseCodeBlock

Execute a piece of code.

Example:

lang: python
code: |
    import random
    # (In PDL, set `result` to the output you wish for your code block.)
    result = random.randint(1, 20)

Attributes:

Name	Type	Description
`lang`	`Annotated[Literal['python', 'command', 'jinja', 'pdl', 'ipython'], BeforeValidator(_ensure_lower)]`	Programming language of the code.
`code`	`BlockType`	Code to execute.

Source code in src/pdl/pdl_ast.py

class CodeBlock(BaseCodeBlock):
    """
    Execute a piece of code.

    Example:
    ```PDL
    lang: python
    code: |
        import random
        # (In PDL, set `result` to the output you wish for your code block.)
        result = random.randint(1, 20)
    ```
    """

    lang: Annotated[
        Literal["python", "command", "jinja", "pdl", "ipython"],
        BeforeValidator(_ensure_lower),
    ]
    """Programming language of the code.
    """
    code: "BlockType"
    """Code to execute.
    """

`lang: Annotated[Literal['python', 'command', 'jinja', 'pdl', 'ipython'], BeforeValidator(_ensure_lower)]` `instance-attribute`

Programming language of the code.

`code: BlockType` `instance-attribute`

Code to execute.

`ArgsBlock`

Bases: BaseCodeBlock

Execute a command line, which will spawn a subprocess with the given argument vector. Note: if you need a shell script execution, you must wrap your command line in /bin/sh or some shell of your choosing.

Example:

args:
- /bin/sh
- "-c"
- "if [[ $x = 1 ]]; then echo y; else echo n; fi"

Attributes:

Name	Type	Description
`args`	`list[ExpressionStr]`	The argument vector to spawn.

Source code in src/pdl/pdl_ast.py

class ArgsBlock(BaseCodeBlock):
    """
    Execute a command line, which will spawn a subprocess with the given argument vector. Note: if you need a shell script execution, you must wrap your command line in /bin/sh or some shell of your choosing.

    Example:
    ```PDL
    args:
    - /bin/sh
    - "-c"
    - "if [[ $x = 1 ]]; then echo y; else echo n; fi"
    ```
    """

    lang: Annotated[Literal["command"], BeforeValidator(_ensure_lower)] = "command"
    args: list[ExpressionStr]
    """The argument vector to spawn.
    """

`args: list[ExpressionStr]` `instance-attribute`

The argument vector to spawn.

`GetBlock`

Bases: LeafBlock

Get the value of a variable.

The GetBlock is deprecated. Use DataBlock instead.

Attributes:

Name	Type	Description
`get`	`str`	Name of the variable to access.

Source code in src/pdl/pdl_ast.py

class GetBlock(LeafBlock):
    """
    Get the value of a variable.

    The GetBlock is deprecated.  Use DataBlock instead.
    """

    kind: Literal[BlockKind.GET] = BlockKind.GET
    get: str
    """Name of the variable to access."""

`get: str` `instance-attribute`

Name of the variable to access.

`DataBlock`

Bases: LeafBlock

Arbitrary value, equivalent to JSON.

Example. As part of a defs section, set numbers to the list [1, 2, 3, 4]:

defs:
  numbers:
    data: [1, 2, 3, 4]

Example. Evaluate ${ TEST.answer } in Jinja, passing the result to a regex parser with capture groups. Set EXTRACTED_GROUND_TRUTH to an object with attribute answer, a string, containing the value of the capture group.

- data: ${ TEST.answer }
  parser:
    regex: "(.|\n)*#### (?P<answer>([0-9])*)\n*"
    spec:
      answer: string
  def: EXTRACTED_GROUND_TRUTH

Attributes:

Name	Type	Description
`data`	`ExpressionType[Any]`	Value defined.
`raw`	`bool`	Do not evaluate expressions inside strings.

Source code in src/pdl/pdl_ast.py

class DataBlock(LeafBlock):
    """
    Arbitrary value, equivalent to JSON.

    Example. As part of a `defs` section, set `numbers` to the list `[1, 2, 3, 4]`:
    ```PDL
    defs:
      numbers:
        data: [1, 2, 3, 4]
    ```

    Example.  Evaluate `${ TEST.answer }` in
    [Jinja](https://jinja.palletsprojects.com/en/stable/), passing
    the result to a regex parser with capture groups.  Set
    `EXTRACTED_GROUND_TRUTH` to an object with attribute `answer`,
    a string, containing the value of the capture group.
    ```PDL
    - data: ${ TEST.answer }
      parser:
        regex: "(.|\\n)*#### (?P<answer>([0-9])*)\\n*"
        spec:
          answer: string
      def: EXTRACTED_GROUND_TRUTH
    ```
    """

    kind: Literal[BlockKind.DATA] = BlockKind.DATA
    data: ExpressionType[Any]
    """Value defined."""
    raw: bool = False
    """Do not evaluate expressions inside strings."""

`data: ExpressionType[Any]` `instance-attribute`

Value defined.

`raw: bool = False` `class-attribute` `instance-attribute`

Do not evaluate expressions inside strings.

`TextBlock`

Bases: StructuredBlock

Create the concatenation of the stringify version of the result of each block of the list of blocks.

Attributes:

Name	Type	Description
`text`	`BlockOrBlocksType`	Body of the text.

Source code in src/pdl/pdl_ast.py

class TextBlock(StructuredBlock):
    """Create the concatenation of the stringify version of the result of each block of the list of blocks."""

    kind: Literal[BlockKind.TEXT] = BlockKind.TEXT
    text: "BlockOrBlocksType"
    """Body of the text.
    """

`text: BlockOrBlocksType` `instance-attribute`

Body of the text.

`LastOfBlock`

Bases: StructuredBlock

Return the value of the last block if the list of blocks.

Attributes:

Name	Type	Description
`lastOf`	`list[BlockType]`	Sequence of blocks to execute.

Source code in src/pdl/pdl_ast.py

class LastOfBlock(StructuredBlock):
    """Return the value of the last block if the list of blocks."""

    kind: Literal[BlockKind.LASTOF] = BlockKind.LASTOF
    lastOf: list["BlockType"]
    """Sequence of blocks to execute."""

`lastOf: list[BlockType]` `instance-attribute`

Sequence of blocks to execute.

`ArrayBlock`

Bases: StructuredBlock

Return the array of values computed by each block of the list of blocks.

Attributes:

Name	Type	Description
`array`	`list[BlockType]`	Elements of the array.

Source code in src/pdl/pdl_ast.py

class ArrayBlock(StructuredBlock):
    """Return the array of values computed by each block of the list of blocks."""

    kind: Literal[BlockKind.ARRAY] = BlockKind.ARRAY
    array: list["BlockType"]
    """Elements of the array."""

`array: list[BlockType]` `instance-attribute`

Elements of the array.

`ObjectBlock`

Bases: StructuredBlock

Return the object where the value of each field is defined by a block. If the body of the object is an array, the resulting object is the union of the objects computed by each element of the array.

Source code in src/pdl/pdl_ast.py

class ObjectBlock(StructuredBlock):
    """Return the object where the value of each field is defined by a block. If the body of the object is an array, the resulting object is the union of the objects computed by each element of the array."""

    kind: Literal[BlockKind.OBJECT] = BlockKind.OBJECT
    object: dict[str, "BlockType"] | list["BlockType"]

`MessageBlock`

Bases: LeafBlock

Create a message.

Attributes:

Name	Type	Description
`content`	`BlockType`	Content of the message.
`name`	`OptionalExpressionStr`	For example, the name of the tool that was invoked, for which this message is the tool response.
`tool_call_id`	`OptionalExpressionStr`	The id of the tool invocation for which this message is the tool response.

Source code in src/pdl/pdl_ast.py

class MessageBlock(LeafBlock):
    """Create a message."""

    kind: Literal[BlockKind.MESSAGE] = BlockKind.MESSAGE
    content: "BlockType"
    """Content of the message."""
    name: OptionalExpressionStr = None
    """For example, the name of the tool that was invoked, for which this message is the tool response."""
    tool_call_id: OptionalExpressionStr = None
    """The id of the tool invocation for which this message is the tool response."""

`content: BlockType` `instance-attribute`

Content of the message.

`name: OptionalExpressionStr = None` `class-attribute` `instance-attribute`

For example, the name of the tool that was invoked, for which this message is the tool response.

`tool_call_id: OptionalExpressionStr = None` `class-attribute` `instance-attribute`

The id of the tool invocation for which this message is the tool response.

`IfBlock`

Bases: StructuredBlock

Conditional control structure.

Example:

defs:
  answer:
    read:
    message: "Enter a number? "
if: ${ (answer | int) == 42 }
then: You won!

Attributes:

Name	Type	Description
`condition`	`ExpressionBool`	Condition.
`then`	`BlockType`	Branch to execute if the condition is true.
`else_`	`OptionalBlockType`	Branch to execute if the condition is false.

Source code in src/pdl/pdl_ast.py

class IfBlock(StructuredBlock):
    """
    Conditional control structure.

    Example:
    ```PDL
    defs:
      answer:
        read:
        message: "Enter a number? "
    if: ${ (answer | int) == 42 }
    then: You won!
    ```
    """

    kind: Literal[BlockKind.IF] = BlockKind.IF
    condition: ExpressionBool = Field(alias="if")
    """Condition.
    """
    then: "BlockType"
    """Branch to execute if the condition is true.
    """
    else_: OptionalBlockType = Field(default=None, alias="else")
    """Branch to execute if the condition is false.
    """

`condition: ExpressionBool = Field(alias='if')` `class-attribute` `instance-attribute`

Condition.

`then: BlockType` `instance-attribute`

Branch to execute if the condition is true.

`else_: OptionalBlockType = Field(default=None, alias='else')` `class-attribute` `instance-attribute`

Branch to execute if the condition is false.

`MatchCase`

Bases: BaseModel

Case of a match.

Attributes:

Name	Type	Description
`case`	`Optional[PatternType]`	Value to match.
`if_`	`OptionalExpressionBool`	Boolean condition to satisfy.
`then`	`BlockType`	Branch to execute if the value is matched and the condition is satisfied.

Source code in src/pdl/pdl_ast.py

class MatchCase(BaseModel):
    """Case of a match."""

    model_config = ConfigDict(extra="forbid")
    case: Optional[PatternType] = None
    """Value to match.
    """
    if_: OptionalExpressionBool = Field(default=None, alias="if")
    """Boolean condition to satisfy.
    """
    then: "BlockType"
    """Branch to execute if the value is matched and the condition is satisfied.
    """
    # Field for internal use
    pdl__case_result: OptionalBool = None
    pdl__if_result: OptionalBool = None
    pdl__matched: OptionalBool = None

`case: Optional[PatternType] = None` `class-attribute` `instance-attribute`

Value to match.

`if_: OptionalExpressionBool = Field(default=None, alias='if')` `class-attribute` `instance-attribute`

Boolean condition to satisfy.

`then: BlockType` `instance-attribute`

Branch to execute if the value is matched and the condition is satisfied.

`MatchBlock`

Bases: StructuredBlock

Match control structure.

Example: ```PDL defs: answer: read: message: "Enter a number? " match: ${ (answer | int) } with: - case: 42 then: You won! - case: any: def: x if: ${ x > 42 } then: Too high - then: Too low

Attributes:

Name	Type	Description
`match_`	`ExpressionType[Any]`	Matched expression.
`with_`	`list[MatchCase]`	List of cases to match.

Source code in src/pdl/pdl_ast.py

class MatchBlock(StructuredBlock):
    """Match control structure.

    Example:
    ```PDL
    defs:
      answer:
        read:
        message: "Enter a number? "
    match: ${ (answer | int) }
    with:
    - case: 42
      then: You won!
    - case:
        any:
        def: x
      if: ${ x > 42 }
      then: Too high
    - then: Too low
    """

    kind: Literal[BlockKind.MATCH] = BlockKind.MATCH
    match_: ExpressionType[Any] = Field(alias="match")
    """Matched expression.
    """
    with_: list[MatchCase] = Field(alias="with")
    """List of cases to match.
    """

`match_: ExpressionType[Any] = Field(alias='match')` `class-attribute` `instance-attribute`

Matched expression.

`with_: list[MatchCase] = Field(alias='with')` `class-attribute` `instance-attribute`

List of cases to match.

`JoinConfig`

Bases: BaseModel

Configure how loop iterations should be combined.

Source code in src/pdl/pdl_ast.py

class JoinConfig(BaseModel):
    """Configure how loop iterations should be combined."""

    model_config = ConfigDict(
        extra="forbid", use_attribute_docstrings=True, validate_by_name=True
    )

`JoinText`

Bases: JoinConfig

Join loop iterations as a string.

Attributes:

Name	Type	Description
`as_`	`Literal['text']`	String concatenation of the result of each iteration.
`with_`	`str`	String used to concatenate each iteration of the loop.

Source code in src/pdl/pdl_ast.py

class JoinText(JoinConfig):
    """Join loop iterations as a string."""

    as_: Literal["text"] = Field(alias="as", default="text")
    """String concatenation of the result of each iteration.
    """

    with_: str = Field(alias="with", default="")
    """String used to concatenate each iteration of the loop.
    """

`as_: Literal['text'] = Field(alias='as', default='text')` `class-attribute` `instance-attribute`

String concatenation of the result of each iteration.

`with_: str = Field(alias='with', default='')` `class-attribute` `instance-attribute`

String used to concatenate each iteration of the loop.

`JoinArray`

Bases: JoinConfig

Join loop iterations as an array.

Attributes:

Name	Type	Description
`as_`	`Literal['array']`	Return the result of each iteration as an array.

Source code in src/pdl/pdl_ast.py

class JoinArray(JoinConfig):
    """Join loop iterations as an array."""

    as_: Literal["array"] = Field(alias="as")
    """Return the result of each iteration as an array.
    """

`as_: Literal['array'] = Field(alias='as')` `class-attribute` `instance-attribute`

Return the result of each iteration as an array.

`JoinObject`

Bases: JoinConfig

Join loop iterations as an object.

Attributes:

Name	Type	Description
`as_`	`Literal['object']`	Return the union of the objects created at each iteration.

Source code in src/pdl/pdl_ast.py

class JoinObject(JoinConfig):
    """Join loop iterations as an object."""

    as_: Literal["object"] = Field(alias="as")
    """Return the union of the objects created at each iteration.
    """

`as_: Literal['object'] = Field(alias='as')` `class-attribute` `instance-attribute`

Return the union of the objects created at each iteration.

`JoinLastOf`

Bases: JoinConfig

Join loop iterations as the value of the last iteration.

Attributes:

Name	Type	Description
`as_`	`Literal['lastOf']`	Return the result of the last iteration.

Source code in src/pdl/pdl_ast.py

class JoinLastOf(JoinConfig):
    """Join loop iterations as the value of the last iteration."""

    as_: Literal["lastOf"] = Field(alias="as")
    """Return the result of the last iteration.
    """

`as_: Literal['lastOf'] = Field(alias='as')` `class-attribute` `instance-attribute`

Return the result of the last iteration.

`JoinReduce`

Bases: JoinConfig

Join loop iterations as the value of the last iteration.

Attributes:

Name	Type	Description
`reduce`	`ExpressionType[Callable]`	Function used to combine the results.

Source code in src/pdl/pdl_ast.py

class JoinReduce(JoinConfig):
    """Join loop iterations as the value of the last iteration."""

    as_: Literal["reduce"] = Field(alias="as", default="reduce")

    reduce: ExpressionType[Callable]
    """Function used to combine the results."""

`reduce: ExpressionType[Callable]` `instance-attribute`

Function used to combine the results.

`RepeatBlock`

Bases: StructuredBlock

Repeat the execution of a block sequentially. The scope and pdl_context are accumulated in between iterations.

For loop example:

for:
    number: [1, 2, 3, 4]
    name: ["Bob", "Carol", "David", "Ernest"]
repeat:
    "${ name }'s number is ${ number }\n"

While loop:

defs:
  i: 0
while: ${i < 5}
repeat:
    defs:
      i: ${ i + 1}
    data: ${ i }
join:
  as: array

Attributes:

Name	Type	Description
`for_`	`Optional[dict[str, ExpressionType[list]]]`	Arrays to iterate over.
`index`	`OptionalStr`	Name of the variable containing the loop iteration.
`while_`	`ExpressionBool`	Condition to stay at the beginning of the loop.
`repeat`	`BlockType`	Body of the loop.
`until`	`ExpressionBool`	Condition to exit at the end of the loop.
`maxIterations`	`OptionalExpressionInt`	Maximal number of iterations to perform.
`join`	`JoinType`	Define how to combine the result of each iteration.

Source code in src/pdl/pdl_ast.py

class RepeatBlock(StructuredBlock):
    """
    Repeat the execution of a block sequentially.
    The scope and `pdl_context` are accumulated in between iterations.

    For loop example:
    ```PDL
    for:
        number: [1, 2, 3, 4]
        name: ["Bob", "Carol", "David", "Ernest"]
    repeat:
        "${ name }'s number is ${ number }\\n"
    ```

    While loop:
    ```PDL
    defs:
      i: 0
    while: ${i < 5}
    repeat:
        defs:
          i: ${ i + 1}
        data: ${ i }
    join:
      as: array
    ```
    """

    kind: Literal[BlockKind.REPEAT] = BlockKind.REPEAT
    for_: Optional[dict[str, ExpressionType[list]]] = Field(default=None, alias="for")
    """Arrays to iterate over.
    """
    index: OptionalStr = None
    """Name of the variable containing the loop iteration.
    """
    while_: ExpressionBool = Field(default=True, alias="while")
    """Condition to stay at the beginning of the loop.
    """
    repeat: "BlockType"
    """Body of the loop.
    """
    until: ExpressionBool = False
    """Condition to exit at the end of the loop.
    """
    maxIterations: OptionalExpressionInt = None
    """Maximal number of iterations to perform.
    """
    join: JoinType = JoinText()
    """Define how to combine the result of each iteration.
    """
    # Field for internal use
    pdl__trace: Optional[list["BlockType"]] = None

`for_: Optional[dict[str, ExpressionType[list]]] = Field(default=None, alias='for')` `class-attribute` `instance-attribute`

Arrays to iterate over.

`index: OptionalStr = None` `class-attribute` `instance-attribute`

Name of the variable containing the loop iteration.

`while_: ExpressionBool = Field(default=True, alias='while')` `class-attribute` `instance-attribute`

Condition to stay at the beginning of the loop.

`repeat: BlockType` `instance-attribute`

Body of the loop.

`until: ExpressionBool = False` `class-attribute` `instance-attribute`

Condition to exit at the end of the loop.

`maxIterations: OptionalExpressionInt = None` `class-attribute` `instance-attribute`

Maximal number of iterations to perform.

`join: JoinType = JoinText()` `class-attribute` `instance-attribute`

Define how to combine the result of each iteration.

`MapBlock`

Bases: StructuredBlock

Independent executions of a block. Repeat the execution of a block starting from the initial scope and pdl_context.

For loop example:

for:
    number: [1, 2, 3, 4]
    name: ["Bob", "Carol", "David", "Ernest"]
map:
    "${ name }'s number is ${ number }\n"

Bounded loop:

index: i
maxIterations: 5
map:
    ${ i }
join:
  as: array

Attributes:

Name	Type	Description
`for_`	`Optional[dict[str, ExpressionType[list]]]`	Arrays to iterate over.
`index`	`OptionalStr`	Name of the variable containing the loop iteration.
`map`	`BlockType`	Body of the iterator.
`maxIterations`	`OptionalExpressionInt`	Maximal number of iterations to perform.
`join`	`JoinType`	Define how to combine the result of each iteration.
`maxWorkers`	`OptionalInt`	Maximal number of workers to execute the map in parallel. Is it is set to `0`, the execution is sequential otherwise it is given as argument to the `ThreadPoolExecutor`.

Source code in src/pdl/pdl_ast.py

class MapBlock(StructuredBlock):
    """
    Independent executions of  a block.
    Repeat the execution of a block starting from the initial scope
    and `pdl_context`.

    For loop example:
    ```PDL
    for:
        number: [1, 2, 3, 4]
        name: ["Bob", "Carol", "David", "Ernest"]
    map:
        "${ name }'s number is ${ number }\\n"
    ```

    Bounded loop:
    ```PDL
    index: i
    maxIterations: 5
    map:
        ${ i }
    join:
      as: array
    ```
    """

    kind: Literal[BlockKind.MAP] = BlockKind.MAP
    for_: Optional[dict[str, ExpressionType[list]]] = Field(default=None, alias="for")
    """Arrays to iterate over.
    """
    index: OptionalStr = None
    """Name of the variable containing the loop iteration.
    """
    map: "BlockType"
    """Body of the iterator.
    """
    maxIterations: OptionalExpressionInt = None
    """Maximal number of iterations to perform.
    """
    join: JoinType = JoinText()
    """Define how to combine the result of each iteration.
    """
    maxWorkers: OptionalInt = None
    """Maximal number of workers to execute the map in parallel. Is it is set to `0`, the execution is sequential otherwise it is given as argument to the `ThreadPoolExecutor`.
    """
    # Field for internal use
    pdl__trace: Optional[list["BlockType"]] = None

`for_: Optional[dict[str, ExpressionType[list]]] = Field(default=None, alias='for')` `class-attribute` `instance-attribute`

Arrays to iterate over.

`index: OptionalStr = None` `class-attribute` `instance-attribute`

Name of the variable containing the loop iteration.

`map: BlockType` `instance-attribute`

Body of the iterator.

`maxIterations: OptionalExpressionInt = None` `class-attribute` `instance-attribute`

Maximal number of iterations to perform.

`join: JoinType = JoinText()` `class-attribute` `instance-attribute`

Define how to combine the result of each iteration.

`maxWorkers: OptionalInt = None` `class-attribute` `instance-attribute`

Maximal number of workers to execute the map in parallel. Is it is set to 0, the execution is sequential otherwise it is given as argument to the ThreadPoolExecutor.

`ReadBlock`

Bases: LeafBlock

Read from a file or standard input.

Example. Read from the standard input with a prompt starting with >.

read:
message: "> "

Example. Read the file ./data.yaml in the same directory of the PDL file containing the block and parse it into YAML.

read: ./data.yaml
parser: yaml

Attributes:

Name	Type	Description
`read`	`OptionalExpressionStr`	Name of the file to read. If `None`, read the standard input.
`message`	`OptionalStr`	Message to prompt the user to enter a value.
`multiline`	`bool`	Indicate if one or multiple lines should be read.

Source code in src/pdl/pdl_ast.py

class ReadBlock(LeafBlock):
    """Read from a file or standard input.

    Example. Read from the standard input with a prompt starting with `> `.
    ```PDL
    read:
    message: "> "
    ```

    Example. Read the file `./data.yaml` in the same directory of the PDL file containing the block and parse it into YAML.
    ```PDL
    read: ./data.yaml
    parser: yaml
    ```
    """

    kind: Literal[BlockKind.READ] = BlockKind.READ
    read: OptionalExpressionStr
    """Name of the file to read. If `None`, read the standard input.
    """
    message: OptionalStr = None
    """Message to prompt the user to enter a value.
    """
    multiline: bool = False
    """Indicate if one or multiple lines should be read.
    """

`read: OptionalExpressionStr` `instance-attribute`

Name of the file to read. If None, read the standard input.

`message: OptionalStr = None` `class-attribute` `instance-attribute`

Message to prompt the user to enter a value.

`multiline: bool = False` `class-attribute` `instance-attribute`

Indicate if one or multiple lines should be read.

`IncludeBlock`

Bases: StructuredBlock

Include a PDL file.

Attributes:

Name	Type	Description
`include`	`str`	Name of the file to include.

Source code in src/pdl/pdl_ast.py

class IncludeBlock(StructuredBlock):
    """Include a PDL file."""

    kind: Literal[BlockKind.INCLUDE] = BlockKind.INCLUDE
    include: str
    """Name of the file to include.
    """
    # Field for internal use
    pdl__trace: OptionalBlockType = None

`include: str` `instance-attribute`

Name of the file to include.

`ImportBlock`

Bases: LeafBlock

Import a PDL file.

Attributes:

Name	Type	Description
`import_`	`str`	Name of the file to import.

Source code in src/pdl/pdl_ast.py

class ImportBlock(LeafBlock):
    """Import a PDL file."""

    kind: Literal[BlockKind.IMPORT] = BlockKind.IMPORT
    import_: str = Field(alias="import")
    """Name of the file to import.
    """
    # Field for internal use
    pdl__trace: OptionalBlockType = None

`import_: str = Field(alias='import')` `class-attribute` `instance-attribute`

Name of the file to import.

`AggregatorConfig`

Bases: BaseModel

Common fields for all aggregator configurations.

Attributes:

Name	Type	Description
`description`	`Optional[str]`	Documentation associated to the aggregator config.

Source code in src/pdl/pdl_ast.py

class AggregatorConfig(BaseModel):
    """Common fields for all aggregator configurations."""

    model_config = ConfigDict(
        extra="forbid",
        use_attribute_docstrings=True,
        arbitrary_types_allowed=True,
    )

    description: Optional[str] = None
    """Documentation associated to the aggregator config.
    """

`description: Optional[str] = None` `class-attribute` `instance-attribute`

Documentation associated to the aggregator config.

`FileAggregatorConfig`

Bases: AggregatorConfig

Attributes:

Name	Type	Description
`file`	`ExpressionType[str]`	Name of the file to which contribute.
`mode`	`ExpressionType[str]`	File opening mode.
`encoding`	`ExpressionType[Optional[str]]`	File encoding.
`prefix`	`ExpressionType[str]`	Prefix to the contributed value.
`suffix`	`ExpressionType[str]`	Suffix to the contributed value.
`flush`	`ExpressionType[bool]`	Whether to forcibly flush the stream.

Source code in src/pdl/pdl_ast.py

class FileAggregatorConfig(AggregatorConfig):
    file: ExpressionType[str]
    """Name of the file to which contribute."""
    mode: ExpressionType[str] = "w"
    """File opening mode."""
    encoding: ExpressionType[Optional[str]] = "utf-8"
    """File encoding."""
    prefix: ExpressionType[str] = ""
    """Prefix to the contributed value."""
    suffix: ExpressionType[str] = "\n"
    """Suffix to the contributed value."""
    flush: ExpressionType[bool] = False
    """Whether to forcibly flush the stream."""

`file: ExpressionType[str]` `instance-attribute`

Name of the file to which contribute.

`mode: ExpressionType[str] = 'w'` `class-attribute` `instance-attribute`

File opening mode.

`encoding: ExpressionType[Optional[str]] = 'utf-8'` `class-attribute` `instance-attribute`

File encoding.

`prefix: ExpressionType[str] = ''` `class-attribute` `instance-attribute`

Prefix to the contributed value.

`suffix: ExpressionType[str] = '\n'` `class-attribute` `instance-attribute`

Suffix to the contributed value.

`flush: ExpressionType[bool] = False` `class-attribute` `instance-attribute`

Whether to forcibly flush the stream.

`AggregatorBlock`

Bases: LeafBlock

Create a new aggregator that can be use in the contribute field.

Source code in src/pdl/pdl_ast.py

class AggregatorBlock(LeafBlock):
    """Create a new aggregator that can be use in the `contribute` field."""

    kind: Literal[BlockKind.AGGREGATOR] = BlockKind.AGGREGATOR
    aggregator: AggregatorType

`ErrorBlock`

Bases: LeafBlock

Block representing an error generated at runtime.

Attributes:

Name	Type	Description
`msg`	`str`	Error message.
`program`	`BlockType`	Block that raised the error.

Source code in src/pdl/pdl_ast.py

class ErrorBlock(LeafBlock):
    """Block representing an error generated at runtime."""

    kind: Literal[BlockKind.ERROR] = BlockKind.ERROR
    msg: str
    """Error message.
    """
    program: "BlockType"
    """Block that raised the error.
    """

`msg: str` `instance-attribute`

Error message.

`program: BlockType` `instance-attribute`

Block that raised the error.

`EmptyBlock`

Bases: LeafBlock

Block without an action. It can contain definitions.

Source code in src/pdl/pdl_ast.py

class EmptyBlock(LeafBlock):
    """Block without an action. It can contain definitions."""

    kind: Literal[BlockKind.EMPTY] = BlockKind.EMPTY

`Program`

Bases: RootModel

Prompt Declaration Language program (PDL)

Attributes:

Name	Type	Description
`root`	`BlockType`	Entry point to parse a PDL program using Pydantic.

Source code in src/pdl/pdl_ast.py

class Program(RootModel):
    """
    Prompt Declaration Language program (PDL)
    """

    root: BlockType
    """Entry point to parse a PDL program using Pydantic.
    """

`root: BlockType` `instance-attribute`

Entry point to parse a PDL program using Pydantic.

`get_default_model_parameters() -> list[dict[str, Any]]`

Model-specific defaults to apply

Source code in src/pdl/pdl_ast.py

def get_default_model_parameters() -> list[dict[str, Any]]:
    """Model-specific defaults to apply"""
    return [
        {
            "*watsonx/*": {
                "temperature": 0,
            },
        },
        {
            "*watsonx_text*": {
                "decoding_method": DECODING_METHOD,
                "max_tokens": MAX_NEW_TOKENS,
                "min_new_tokens": MIN_NEW_TOKENS,
                "repetition_penalty": REPETITION_PENALTY,
            },
        },
        # Note that Replicate may no longer support granite 3.0
        {
            "*granite-3.0*": {
                "temperature": 0,
                "roles": {
                    "system": {
                        "pre_message": "<|start_of_role|>system<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "user": {
                        "pre_message": "<|start_of_role|>user<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "assistant": {
                        "pre_message": "<|start_of_role|>assistant<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "available_tools": {
                        "pre_message": "<|start_of_role|>available_tools<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "tool_response": {
                        "pre_message": "<|start_of_role|>tool_response<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                },
                "final_prompt_value": "<|start_of_role|>assistant<|end_of_role|>",
            }
        },
        # Note that we match both granite-3.0 and 3.1 rather than using a granite-3.* wildcard
        {
            "*granite-3.1*": {
                "temperature": 0,
                "roles": {
                    "system": {
                        "pre_message": "<|start_of_role|>system<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "user": {
                        "pre_message": "<|start_of_role|>user<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "assistant": {
                        "pre_message": "<|start_of_role|>assistant<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "tools": {
                        "pre_message": "<|start_of_role|>tools<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "tool_response": {
                        "pre_message": "<|start_of_role|>tool_response<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "documents": {
                        "pre_message": "<|start_of_role|>documents<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                },
                "final_prompt_value": "<|start_of_role|>assistant<|end_of_role|>",
            }
        },
        {
            "*granite-3.2*": {
                "temperature": 0,
                "roles": {
                    "system": {
                        "pre_message": "<|start_of_role|>system<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "user": {
                        "pre_message": "<|start_of_role|>user<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "assistant": {
                        "pre_message": "<|start_of_role|>assistant<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "tools": {
                        "pre_message": "<|start_of_role|>tools<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "tool_response": {
                        "pre_message": "<|start_of_role|>tool_response<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                    "documents": {
                        "pre_message": "<|start_of_role|>documents<|end_of_role|>",
                        "post_message": "<|end_of_text|>",
                    },
                },
                "final_prompt_value": "<|start_of_role|>assistant<|end_of_role|>",
            }
        },
        # models on Ollama (e.g. granite-code, granite3-dense, granite3.1-dense)
        {
            "ollama/*": {
                "temperature": 0,
            },
        },
        {
            "ollama_chat/*": {
                "temperature": 0,
            },
        },
    ]

`get_sampling_defaults() -> list[dict[str, Any]]`

Model-specific defaults to apply if we are sampling.

Source code in src/pdl/pdl_ast.py

def get_sampling_defaults() -> list[dict[str, Any]]:
    """Model-specific defaults to apply if we are sampling."""
    return [
        {
            "*": {
                "temperature": TEMPERATURE_SAMPLING,
                "top_k": TOP_K_SAMPLING,
                "top_p": TOP_P_SAMPLING,
            }
        }
    ]

Interpreter

Classes:

Name	Description
`InterpreterConfig`	Configuration parameters of the PDL interpreter.

Functions:

Name	Description
`exec_program`	Execute a PDL program given as a value of type `pdl.pdl_ast.Program`.
`exec_dict`	Execute a PDL program given as a dictionary.
`exec_str`	Execute a PDL program given as YAML string.
`exec_file`	Execute a PDL program given as YAML file.

`InterpreterConfig`

Bases: TypedDict

Configuration parameters of the PDL interpreter.

Attributes:

Name	Type	Description
`yield_result`	`bool`	Print incrementally result of the execution.
`yield_background`	`bool`	Print the program background messages during the execution.
`batch`	`int`	Model inference mode:
`role`	`RoleType`	Default role.
`cwd`	`Path`	Path considered as the current working directory for file reading.
`replay`	`dict[str, Any]`	Execute the program reusing some already computed values.
`event_loop`	`AbstractEventLoop`	Event loop to schedule LLM calls.

Source code in src/pdl/pdl.py

class InterpreterConfig(TypedDict, total=False):
    """Configuration parameters of the PDL interpreter."""

    yield_result: bool
    """Print incrementally result of the execution.
    """
    yield_background: bool
    """Print the program background messages during the execution.
    """
    batch: int
    """Model inference mode:
         - 0: streaming
         - 1: non-streaming
    """
    role: RoleType
    """Default role.
    """
    cwd: Path
    """Path considered as the current working directory for file reading."""
    replay: dict[str, Any]
    """Execute the program reusing some already computed values.
    """
    event_loop: AbstractEventLoop
    """Event loop to schedule LLM calls."""

`yield_result: bool` `instance-attribute`

Print incrementally result of the execution.

`yield_background: bool` `instance-attribute`

Print the program background messages during the execution.

`batch: int` `instance-attribute`

Model inference mode: - 0: streaming - 1: non-streaming

`role: RoleType` `instance-attribute`

Default role.

`cwd: Path` `instance-attribute`

Path considered as the current working directory for file reading.

`replay: dict[str, Any]` `instance-attribute`

Execute the program reusing some already computed values.

`event_loop: AbstractEventLoop` `instance-attribute`

Event loop to schedule LLM calls.

`exec_program(prog: Program, config: Optional[InterpreterConfig] = None, scope: Optional[ScopeType | dict[str, Any]] = None, loc: Optional[PdlLocationType] = None, output: Literal['result', 'all'] = 'result') -> Any`

Execute a PDL program given as a value of type pdl.pdl_ast.Program.

Parameters:

Name	Type	Description	Default
`prog`	`Program`	Program to execute.	required
`config`	`Optional[InterpreterConfig]`	Interpreter configuration. Defaults to None.	`None`
`scope`	`Optional[ScopeType \| dict[str, Any]]`	Environment defining the initial variables in scope to execute the program. Defaults to None.	`None`
`loc`	`Optional[PdlLocationType]`	Source code location mapping. Defaults to None.	`None`
`output`	`Literal['result', 'all']`	Configure the output of the returned value of this function. Defaults to `"result"`	`'result'`

Returns:

Type	Description
`Any`	Return the final result if `output` is set to `"result"`. If set of `all`, it returns a dictionary containing, `result`, `scope`, `trace`, and `replay`.

Source code in src/pdl/pdl.py

def exec_program(
    prog: Program,
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    loc: Optional[PdlLocationType] = None,
    output: Literal["result", "all"] = "result",
) -> Any:
    """Execute a PDL program given as a value of type `pdl.pdl_ast.Program`.

    Args:
        prog: Program to execute.
        config: Interpreter configuration. Defaults to None.
        scope: Environment defining the initial variables in scope to execute the program. Defaults to None.
        loc: Source code location mapping. Defaults to None.
        output: Configure the output of the returned value of this function. Defaults to `"result"`

    Returns:
        Return the final result if `output` is set to `"result"`. If set of `all`, it returns a dictionary containing, `result`, `scope`, `trace`, and `replay`.
    """
    config = config or InterpreterConfig()
    config["replay"] = dict(config.get("replay", {}))
    state = InterpreterState(**config)
    if not isinstance(scope, PdlDict):
        scope = PdlDict(scope or {})
    loc = loc or empty_block_location
    initial_scope = {"pdl_model_default_parameters": get_default_model_parameters()}
    future_result, _, future_scope, trace = process_prog(
        state, scope | initial_scope, prog, loc
    )
    result = future_result.result()
    match output:
        case "result":
            return result
        case "all":
            scope = future_scope.result()
            return {
                "result": result,
                "scope": scope,
                "trace": trace,
                "replay": state.replay,
            }
        case _:
            assert False, 'The `output` variable should be "result" or "all"'

`exec_dict(prog: dict[str, Any], config: Optional[InterpreterConfig] = None, scope: Optional[ScopeType | dict[str, Any]] = None, loc: Optional[PdlLocationType] = None, output: Literal['result', 'all'] = 'result') -> Any`

Execute a PDL program given as a dictionary.

Parameters:

Name	Type	Description	Default
`prog`	`dict[str, Any]`	Program to execute.	required
`config`	`Optional[InterpreterConfig]`	Interpreter configuration. Defaults to None.	`None`
`scope`	`Optional[ScopeType \| dict[str, Any]]`	Environment defining the initial variables in scope to execute the program. Defaults to None.	`None`
`loc`	`Optional[PdlLocationType]`	Source code location mapping. Defaults to None.	`None`
`output`	`Literal['result', 'all']`	Configure the output of the returned value of this function. Defaults to `"result"`	`'result'`

Returns:

Type	Description
`Any`	Return the final result.

Source code in src/pdl/pdl.py

def exec_dict(
    prog: dict[str, Any],
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    loc: Optional[PdlLocationType] = None,
    output: Literal["result", "all"] = "result",
) -> Any:
    """Execute a PDL program given as a dictionary.

    Args:
        prog: Program to execute.
        config: Interpreter configuration. Defaults to None.
        scope: Environment defining the initial variables in scope to execute the program. Defaults to None.
        loc: Source code location mapping. Defaults to None.
        output: Configure the output of the returned value of this function. Defaults to `"result"`

    Returns:
        Return the final result.
    """
    program = parse_dict(prog)
    result = exec_program(program, config, scope, loc, output)
    return result

`exec_str(prog: str, config: Optional[InterpreterConfig] = None, scope: Optional[ScopeType | dict[str, Any]] = None, output: Literal['result', 'all'] = 'result') -> Any`

Execute a PDL program given as YAML string.

Parameters:

Name	Type	Description	Default
`prog`	`str`	Program to execute.	required
`config`	`Optional[InterpreterConfig]`	Interpreter configuration. Defaults to None.	`None`
`scope`	`Optional[ScopeType \| dict[str, Any]]`	Environment defining the initial variables in scope to execute the program. Defaults to None.	`None`
`output`	`Literal['result', 'all']`	Configure the output of the returned value of this function. Defaults to `"result"`	`'result'`

Returns:

Type	Description
`Any`	Return the final result.

Source code in src/pdl/pdl.py

def exec_str(
    prog: str,
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    output: Literal["result", "all"] = "result",
) -> Any:
    """Execute a PDL program given as YAML string.

    Args:
        prog: Program to execute.
        config: Interpreter configuration. Defaults to None.
        scope: Environment defining the initial variables in scope to execute the program. Defaults to None.
        output: Configure the output of the returned value of this function. Defaults to `"result"`

    Returns:
        Return the final result.
    """
    program, loc = parse_str(prog)
    result = exec_program(program, config, scope, loc, output)
    return result

`exec_file(prog: str | Path, config: Optional[InterpreterConfig] = None, scope: Optional[ScopeType | dict[str, Any]] = None, output: Literal['result', 'all'] = 'result') -> Any`

Execute a PDL program given as YAML file.

Parameters:

Name	Type	Description	Default
`prog`	`str \| Path`	Program to execute.	required
`config`	`Optional[InterpreterConfig]`	Interpreter configuration. Defaults to None.	`None`
`scope`	`Optional[ScopeType \| dict[str, Any]]`	Environment defining the initial variables in scope to execute the program. Defaults to None.	`None`
`output`	`Literal['result', 'all']`	Configure the output of the returned value of this function. Defaults to `"result"`	`'result'`

Returns:

Type	Description
`Any`	Return the final result.

Source code in src/pdl/pdl.py

def exec_file(
    prog: str | Path,
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    output: Literal["result", "all"] = "result",
) -> Any:
    """Execute a PDL program given as YAML file.

    Args:
        prog: Program to execute.
        config: Interpreter configuration. Defaults to None.
        scope: Environment defining the initial variables in scope to execute the program. Defaults to None.
        output: Configure the output of the returned value of this function. Defaults to `"result"`

    Returns:
        Return the final result.
    """
    program, loc = parse_file(prog)
    if config is None:
        config = InterpreterConfig()
    if config.get("cwd") is None:
        config["cwd"] = Path(prog).parent
    result = exec_program(program, config, scope, loc, output)
    return result

API Reference

Program

OptionalStr = TypeAliasType('OptionalStr', Optional[str]) module-attribute

OptionalInt = TypeAliasType('OptionalInt', Optional[int]) module-attribute

OptionalBool = TypeAliasType('OptionalBool', Optional[bool]) module-attribute

OptionalBoolOrStr = TypeAliasType('OptionalBoolOrStr', Optional[Union[bool, str]]) module-attribute

OptionalAny = TypeAliasType('OptionalAny', Optional[Any]) module-attribute

OptionalBlockType = TypeAliasType('OptionalBlockType', Optional['BlockType']) module-attribute

ScopeType = TypeAliasType('ScopeType', PdlDict[str, Any]) module-attribute

ModelInput = TypeAliasType('ModelInput', Sequence[Mapping[str, Any]]) module-attribute

OptionalModelInput = TypeAliasType('OptionalModelInput', Optional[ModelInput]) module-attribute

OptionalPdlLocationType = TypeAliasType('OptionalPdlLocationType', Optional[PdlLocationType]) module-attribute

ExpressionType: TypeAlias = LocalizedExpression[ExpressionTypeT] | ExpressionTypeT | str module-attribute

ExpressionStr = TypeAliasType('ExpressionStr', ExpressionType[str]) module-attribute

OptionalExpressionStr = TypeAliasType('OptionalExpressionStr', Optional[ExpressionStr]) module-attribute

ExpressionInt = TypeAliasType('ExpressionInt', ExpressionType[int]) module-attribute

OptionalExpressionInt = TypeAliasType('OptionalExpressionInt', Optional[ExpressionInt]) module-attribute

ExpressionBool = TypeAliasType('ExpressionBool', ExpressionType[bool]) module-attribute

OptionalExpressionBool = TypeAliasType('OptionalExpressionBool', Optional[ExpressionBool]) module-attribute

PatternType = TypeAliasType('PatternType', Union[None, bool, int, float, str, OrPattern, ArrayPattern, ObjectPattern, AnyPattern]) module-attribute

BasePdlType: TypeAlias = Literal['null', 'boolean', 'string', 'number', 'integer', 'array', 'object', 'bool', 'str', 'float', 'int', 'list', 'obj'] module-attribute

PdlTypeType = TypeAliasType('PdlTypeType', Annotated["Union[None, BasePdlType, EnumPdlType, list['PdlTypeType'], OptionalPdlType, JsonSchemaTypePdlType, ObjectPdlType, dict[str, 'PdlTypeType']]", Field(union_mode='left_to_right')]) module-attribute

ParserType = TypeAliasType('ParserType', Union[Literal['json', 'jsonl', 'yaml'], PdlParser, RegexParser]) module-attribute

OptionalParserType = TypeAliasType('OptionalParserType', Optional[ParserType]) module-attribute

RoleType: TypeAlias = OptionalStr module-attribute

ContributeElement = TypeAliasType('ContributeElement', Union[ContributeTarget, str, dict[str, ContributeValue]]) module-attribute

ExpectationsType = TypeAliasType('ExpectationsType', Sequence[ExpectationType]) module-attribute

OptionalPdlTiming = TypeAliasType('OptionalPdlTiming', Optional[PdlTiming]) module-attribute

OptionalPdlUsage = TypeAliasType('OptionalPdlUsage', Optional[PdlUsage]) module-attribute

JoinType = TypeAliasType('JoinType', Union[JoinText, JoinArray, JoinObject, JoinLastOf, JoinReduce]) module-attribute

BlockType = TypeAliasType('BlockType', Union[None, bool, int, float, str, AdvancedBlockType]) module-attribute

BlockOrBlocksType: TypeAlias = BlockType | list[BlockType] module-attribute

PdlLocationType

LocalizedExpression

Pattern

def_: OptionalStr = Field(default=None, alias='def') class-attribute instance-attribute

OrPattern

anyOf: list[PatternType] instance-attribute

ArrayPattern

array: list[PatternType] instance-attribute

ObjectPattern

object: dict[str, PatternType] instance-attribute

AnyPattern

PdlType

OptionalPdlType

optional: PdlTypeType instance-attribute

JsonSchemaTypePdlType

type: str | list[str] instance-attribute

EnumPdlType

enum: list[Any] instance-attribute

ObjectPdlType

object: dict[str, PdlTypeType] instance-attribute

Parser

description: OptionalStr = None class-attribute instance-attribute

spec: PdlTypeType = None class-attribute instance-attribute

PdlParser

pdl: BlockType instance-attribute

RegexParser

regex: str instance-attribute

mode: Annotated[Literal['search', 'match', 'fullmatch', 'split', 'findall'], BeforeValidator(_ensure_lower)] = 'fullmatch' class-attribute instance-attribute

ContributeTarget

ContributeValue

value: ExpressionType[Any] instance-attribute

ExpectationType

expect: ExpressionType instance-attribute

feedback: Optional[ExpressionType[FunctionBlock]] = None class-attribute instance-attribute

PdlTiming

start_nanos: OptionalInt = 0 class-attribute instance-attribute

end_nanos: OptionalInt = 0 class-attribute instance-attribute

first_use_nanos: OptionalInt = 0 class-attribute instance-attribute

timezone: OptionalStr = '' class-attribute instance-attribute

PdlUsage

completion_tokens: OptionalInt = 0 class-attribute instance-attribute

prompt_tokens: OptionalInt = 0 class-attribute instance-attribute

Block

description: OptionalStr = None class-attribute instance-attribute

spec: PdlTypeType = None class-attribute instance-attribute

defs: dict[str, BlockType] = {} class-attribute instance-attribute

def_: OptionalStr = Field(default=None, alias='def') class-attribute instance-attribute

contribute: Sequence[ContributeElement] = [ContributeTarget.RESULT, ContributeTarget.CONTEXT] class-attribute instance-attribute

`OptionalStr = TypeAliasType('OptionalStr', Optional[str])` `module-attribute`

`OptionalInt = TypeAliasType('OptionalInt', Optional[int])` `module-attribute`

`OptionalBool = TypeAliasType('OptionalBool', Optional[bool])` `module-attribute`

`OptionalBoolOrStr = TypeAliasType('OptionalBoolOrStr', Optional[Union[bool, str]])` `module-attribute`

`OptionalAny = TypeAliasType('OptionalAny', Optional[Any])` `module-attribute`

`OptionalBlockType = TypeAliasType('OptionalBlockType', Optional['BlockType'])` `module-attribute`

`ScopeType = TypeAliasType('ScopeType', PdlDict[str, Any])` `module-attribute`

`ModelInput = TypeAliasType('ModelInput', Sequence[Mapping[str, Any]])` `module-attribute`

`OptionalModelInput = TypeAliasType('OptionalModelInput', Optional[ModelInput])` `module-attribute`

`OptionalPdlLocationType = TypeAliasType('OptionalPdlLocationType', Optional[PdlLocationType])` `module-attribute`

`ExpressionType: TypeAlias = LocalizedExpression[ExpressionTypeT] | ExpressionTypeT | str` `module-attribute`

`ExpressionStr = TypeAliasType('ExpressionStr', ExpressionType[str])` `module-attribute`

`OptionalExpressionStr = TypeAliasType('OptionalExpressionStr', Optional[ExpressionStr])` `module-attribute`

`ExpressionInt = TypeAliasType('ExpressionInt', ExpressionType[int])` `module-attribute`

`OptionalExpressionInt = TypeAliasType('OptionalExpressionInt', Optional[ExpressionInt])` `module-attribute`

`ExpressionBool = TypeAliasType('ExpressionBool', ExpressionType[bool])` `module-attribute`

`OptionalExpressionBool = TypeAliasType('OptionalExpressionBool', Optional[ExpressionBool])` `module-attribute`

`PatternType = TypeAliasType('PatternType', Union[None, bool, int, float, str, OrPattern, ArrayPattern, ObjectPattern, AnyPattern])` `module-attribute`

`BasePdlType: TypeAlias = Literal['null', 'boolean', 'string', 'number', 'integer', 'array', 'object', 'bool', 'str', 'float', 'int', 'list', 'obj']` `module-attribute`

`PdlTypeType = TypeAliasType('PdlTypeType', Annotated["Union[None, BasePdlType, EnumPdlType, list['PdlTypeType'], OptionalPdlType, JsonSchemaTypePdlType, ObjectPdlType, dict[str, 'PdlTypeType']]", Field(union_mode='left_to_right')])` `module-attribute`

`ParserType = TypeAliasType('ParserType', Union[Literal['json', 'jsonl', 'yaml'], PdlParser, RegexParser])` `module-attribute`

`OptionalParserType = TypeAliasType('OptionalParserType', Optional[ParserType])` `module-attribute`

`RoleType: TypeAlias = OptionalStr` `module-attribute`

`ContributeElement = TypeAliasType('ContributeElement', Union[ContributeTarget, str, dict[str, ContributeValue]])` `module-attribute`

`ExpectationsType = TypeAliasType('ExpectationsType', Sequence[ExpectationType])` `module-attribute`

`OptionalPdlTiming = TypeAliasType('OptionalPdlTiming', Optional[PdlTiming])` `module-attribute`

`OptionalPdlUsage = TypeAliasType('OptionalPdlUsage', Optional[PdlUsage])` `module-attribute`

`JoinType = TypeAliasType('JoinType', Union[JoinText, JoinArray, JoinObject, JoinLastOf, JoinReduce])` `module-attribute`

`BlockType = TypeAliasType('BlockType', Union[None, bool, int, float, str, AdvancedBlockType])` `module-attribute`

`BlockOrBlocksType: TypeAlias = BlockType | list[BlockType]` `module-attribute`

`PdlLocationType`

`LocalizedExpression`

`Pattern`

`def_: OptionalStr = Field(default=None, alias='def')` `class-attribute` `instance-attribute`

`OrPattern`

`anyOf: list[PatternType]` `instance-attribute`

`ArrayPattern`

`array: list[PatternType]` `instance-attribute`

`ObjectPattern`

`object: dict[str, PatternType]` `instance-attribute`

`AnyPattern`

`PdlType`

`OptionalPdlType`

`optional: PdlTypeType` `instance-attribute`

`JsonSchemaTypePdlType`

`type: str | list[str]` `instance-attribute`

`EnumPdlType`

`enum: list[Any]` `instance-attribute`

`ObjectPdlType`

`object: dict[str, PdlTypeType]` `instance-attribute`

`Parser`

`description: OptionalStr = None` `class-attribute` `instance-attribute`

`spec: PdlTypeType = None` `class-attribute` `instance-attribute`

`PdlParser`

`pdl: BlockType` `instance-attribute`

`RegexParser`

`regex: str` `instance-attribute`

`mode: Annotated[Literal['search', 'match', 'fullmatch', 'split', 'findall'], BeforeValidator(_ensure_lower)] = 'fullmatch'` `class-attribute` `instance-attribute`

`ContributeTarget`

`ContributeValue`

`value: ExpressionType[Any]` `instance-attribute`

`ExpectationType`

`expect: ExpressionType` `instance-attribute`

`feedback: Optional[ExpressionType[FunctionBlock]] = None` `class-attribute` `instance-attribute`

`PdlTiming`

`start_nanos: OptionalInt = 0` `class-attribute` `instance-attribute`

`end_nanos: OptionalInt = 0` `class-attribute` `instance-attribute`

`first_use_nanos: OptionalInt = 0` `class-attribute` `instance-attribute`

`timezone: OptionalStr = ''` `class-attribute` `instance-attribute`

`PdlUsage`

`completion_tokens: OptionalInt = 0` `class-attribute` `instance-attribute`

`prompt_tokens: OptionalInt = 0` `class-attribute` `instance-attribute`

`Block`

`description: OptionalStr = None` `class-attribute` `instance-attribute`

`spec: PdlTypeType = None` `class-attribute` `instance-attribute`

`defs: dict[str, BlockType] = {}` `class-attribute` `instance-attribute`

`def_: OptionalStr = Field(default=None, alias='def')` `class-attribute` `instance-attribute`

`contribute: Sequence[ContributeElement] = [ContributeTarget.RESULT, ContributeTarget.CONTEXT]` `class-attribute` `instance-attribute`

`parser: Annotated[OptionalParserType, BeforeValidator(_ensure_lower)] = None` `class-attribute` `instance-attribute`

`fallback: OptionalBlockType = None` `class-attribute` `instance-attribute`