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.
LeafBlock	Base class for blocks that directly contribute their value to the context.
IndependentEnum	Enumeration for context execution mode in structured blocks.
StructuredBlock	Base class for blocks that do not directly contribute to the context but contain sub-blocks that can contribute.
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.
BaseCodeBlock	Base class for code execution blocks.
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.
JoinConfig	Configure how loop iterations or sequence of blocks should be combined.
JoinText	Join loop iterations or sequence of blocks as a string.
JoinArray	Join loop iterations or sequence of blocks as an array.
JoinObject	Join loop iterations or sequence of blocks as an object.
JoinLastOf	Join loop iterations or sequence of blocks as the value of the last iteration.
JoinReduce	Join loop iterations or sequence of blocks as the value of the last iteration.
SequenceBlock	Generalization of the text, lastOf, array, and object blocks combining a list of blocks with a join operator.
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.
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)
PdlBlock	Wrapper class used to generate proper JSON Schema for PDL blocks.

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.
ModelInput		Type of the input of an LLM call.
OptionalModelInput		Optional value of type ModelInput.
OptionalPdlLocationType		Optional location type.
LocalizedExpressionT		Type variable for the result type of a localized expression.
ExpressionTypeT		Type variable for the result type of an expression.
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.
ExpressionList		Expression evaluating into a list.
OptionalExpressionList		Optional expression evaluating into a list.
ExpressionDictStr		"Expression evaluating into a dict[str, Any].
OptionalExpressionDictStr		Optional expression evaluating into a dict[str, Any].
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 or sequence of blocks.
ExpressionBlock		Expression as blocks
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.

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', PdlLocationType | None) module-attribute

Optional location type.

LocalizedExpressionT = TypeVar('LocalizedExpressionT') module-attribute

Type variable for the result type of a localized expression.

ExpressionTypeT = TypeVar('ExpressionTypeT') module-attribute

Type variable for the result type of an expression.

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', ExpressionStr | None) module-attribute

Optional expression evaluating into a string.

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

Expression evaluating into an int.

OptionalExpressionInt = TypeAliasType('OptionalExpressionInt', ExpressionInt | None) module-attribute

Optional expression evaluating into an int.

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

Expression evaluating into a bool.

OptionalExpressionBool = TypeAliasType('OptionalExpressionBool', ExpressionBool | None) module-attribute

Optional expression evaluating into a bool.

ExpressionList = TypeAliasType('ExpressionList', ExpressionType[list]) module-attribute

Expression evaluating into a list.

OptionalExpressionList = TypeAliasType('OptionalExpressionList', ExpressionList | None) module-attribute

Optional expression evaluating into a list.

ExpressionDictStr = TypeAliasType('ExpressionDictStr', ExpressionType[dict[str, Any]]) module-attribute

"Expression evaluating into a dict[str, Any].

OptionalExpressionDictStr = TypeAliasType('OptionalExpressionDictStr', ExpressionDictStr | None) module-attribute

Optional expression evaluating into a dict[str, Any].

PatternType = TypeAliasType('PatternType', 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', 'csv'], 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 or sequence of blocks.

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

Expression as blocks

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

Different types of structured blocks.

BlockType = TypeAliasType('BlockType', Union[ExpressionBlock, 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: LocalizedExpressionT | None = 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.

Attributes:

Name	Type	Description
any	Literal[None]	Matches any value (set to None to indicate wildcard matching).

Source code in src/pdl/pdl_ast.py

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

    any: Literal[None]
    """Matches any value (set to None to indicate wildcard matching)."""

any: Literal[None] instance-attribute

Matches any value (set to None to indicate wildcard matching).

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	The wrapped type that is optional.

Source code in src/pdl/pdl_ast.py

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

    optional: "PdlTypeType"
    """The wrapped type that is optional."""

optional: PdlTypeType instance-attribute

The wrapped type that is optional.

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[Any]	English description of the expectation
feedback	ExpressionType[FunctionBlock] | None	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[Any]
    """English description of the expectation"""

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

expect: ExpressionType[Any] instance-attribute

English description of the expectation

feedback: ExpressionType[FunctionBlock] | None = 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
model_calls	int	Number of calls to LLMs.
completion_tokens	int	Completion tokens consumed.
prompt_tokens	int	Prompt tokens consumed.

Source code in src/pdl/pdl_ast.py

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

    model_calls: int = 0
    """Number of calls to LLMs.
    """
    completion_tokens: int = 0
    """Completion tokens consumed.
    """
    prompt_tokens: int = 0
    """Prompt tokens consumed.
    """

model_calls: int = 0 class-attribute instance-attribute

Number of calls to LLMs.

completion_tokens: int = 0 class-attribute instance-attribute

Completion tokens consumed.

prompt_tokens: int = 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.

LeafBlock

Bases: Block

Base class for blocks that directly contribute their value to the context.

Source code in src/pdl/pdl_ast.py

class LeafBlock(Block):
    """Base class for blocks that directly contribute their value to the context."""

    # Field for internal use
    pdl__is_leaf: Literal[True] = True

IndependentEnum

Bases: StrEnum

Enumeration for context execution mode in structured blocks.

• INDEPENDENT: Execute with fresh context (parallel execution)
• DEPENDENT: Execute with accumulated context (sequential execution)

Source code in src/pdl/pdl_ast.py

class IndependentEnum(StrEnum):
    """Enumeration for context execution mode in structured blocks.

    - INDEPENDENT: Execute with fresh context (parallel execution)
    - DEPENDENT: Execute with accumulated context (sequential execution)
    """

    INDEPENDENT = "independent"
    DEPENDENT = "dependent"

StructuredBlock

Bases: Block

Base class for blocks that do not directly contribute to the context but contain sub-blocks that can contribute.

Source code in src/pdl/pdl_ast.py

class StructuredBlock(Block):
    """Base class for blocks that do not directly contribute to the context but contain sub-blocks that can contribute."""

    # Field for internal use
    pdl__is_leaf: Literal[False] = False
    context: IndependentEnum = IndependentEnum.DEPENDENT

FunctionBlock

Bases: LeafBlock

Function declaration.

Attributes:

Name	Type	Description
function	dict[str, PdlTypeType] | None	Functions parameters with their types.
return_	BlockType	Body of the function.
signature	Json | None	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: dict[str, PdlTypeType] | None
    """Functions parameters with their types.
    """
    return_: "BlockType" = Field(..., alias="return")
    """Body of the function.
    """
    signature: Json | None = None
    """Function signature computed from the function definition.
    """

function: dict[str, PdlTypeType] | None instance-attribute

Functions parameters with their types.

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

Body of the function.

signature: Json | None = 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[Any]	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[Any] = {}
    """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[Any] = {} 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	float | str | None	Timeout in seconds for completion requests (Defaults to 600 seconds).
temperature	float | str | None	The temperature parameter for controlling the randomness of the output (default is 1.0).
top_p	float | str | None	The top-p parameter for nucleus sampling (default is 1.0).
n	int | str | None	The number of completions to generate (default is 1).
stop	str | list[str] | None	Up to 4 sequences where the LLM API will stop generating further tokens.
max_tokens	int | str | None	The maximum number of tokens in the generated completion (default is infinity).
presence_penalty	float | str | None	It is used to penalize new tokens based on their existence in the text so far.
frequency_penalty	float | str | None	It is used to penalize new tokens based on their frequency in the text so far.
logit_bias	dict | str | None	Used to modify the probability of specific tokens appearing in the completion.
user	str | None	A unique identifier representing your end-user. This can help the LLM provider to monitor and detect abuse.
logprobs	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	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	bool | str | None	Whether to enable parallel execution of tool calls.
extra_headers	dict | str | None	Additional headers to include in the request.
functions	list | str | None	A list of functions to apply to the conversation messages (default is an empty list)
function_call	str | None	The name of the function to call within the conversation (default is an empty string)
base_url	str | None	Base URL for the API (default is None).
api_version	str | None	API version (default is None).
api_key	str | None	API key (default is None).
model_list	list | str | None	List of api base, version, keys.
mock_response	str | None	If provided, return a mock completion response for testing or debugging purposes (default is None).
custom_llm_provider	str | None	Used for Non-OpenAI LLMs, Example usage for bedrock, set model="amazon.titan-tg1-large" and custom_llm_provider="bedrock"
max_retries	int | str | None	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: float | str | None = None
    """Timeout in seconds for completion requests (Defaults to 600 seconds).
    """
    temperature: float | str | None = None
    """The temperature parameter for controlling the randomness of the output (default is 1.0).
    """
    top_p: float | str | None = None
    """The top-p parameter for nucleus sampling (default is 1.0).
    """
    n: int | str | None = 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: str | list[str] | None = None
    """Up to 4 sequences where the LLM API will stop generating further tokens.
    """
    max_tokens: int | str | None = None
    """The maximum number of tokens in the generated completion (default is infinity).
    """
    presence_penalty: float | str | None = None
    """It is used to penalize new tokens based on their existence in the text so far.
    """
    frequency_penalty: float | str | None = None
    """It is used to penalize new tokens based on their frequency in the text so far.
    """
    logit_bias: dict | str | None = None
    """Used to modify the probability of specific tokens appearing in the completion.
    """
    user: str | None = 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: dict | str | None = None
    seed: int | str | None = None
    tools: list | str | None = None
    tool_choice: str | dict | None = None
    logprobs: bool | str | None = 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: int | str | None = 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: bool | str | None = None
    """Whether to enable parallel execution of tool calls."""
    # deployment_id = None
    extra_headers: dict | str | None = None
    """Additional headers to include in the request.
    """
    # soon to be deprecated params by OpenAI
    functions: list | str | None = None
    """A list of functions to apply to the conversation messages (default is an empty list)
    """
    function_call: str | None = 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: str | None = environ.get("OPENAI_BASE_URL")
    """Base URL for the API (default is None).
    """
    api_version: str | None = None
    """API version (default is None).
    """
    api_key: str | None = None
    """API key (default is None).
    """
    model_list: list | str | None = None  # pass in a list of api_base,keys, etc.
    """List of api base, version, keys.
    """
    # Optional liteLLM function params
    mock_response: str | None = None
    """If provided, return a mock completion response for testing or debugging purposes (default is None).
    """
    custom_llm_provider: str | None = None
    """Used for Non-OpenAI LLMs, Example usage for bedrock, set model="amazon.titan-tg1-large" and custom_llm_provider="bedrock"
    """
    max_retries: int | str | None = None
    """The number of retries to attempt (default is 0).
    """

timeout: float | str | None = None class-attribute instance-attribute

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

temperature: float | str | None = None class-attribute instance-attribute

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

top_p: float | str | None = None class-attribute instance-attribute

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

n: int | str | None = None class-attribute instance-attribute

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

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

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

max_tokens: int | str | None = None class-attribute instance-attribute

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

presence_penalty: float | str | None = None class-attribute instance-attribute

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

frequency_penalty: float | str | None = None class-attribute instance-attribute

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

logit_bias: dict | str | None = None class-attribute instance-attribute

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

user: str | None = None class-attribute instance-attribute

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

logprobs: bool | str | None = 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: int | str | None = 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.

parallel_tool_calls: bool | str | None = None class-attribute instance-attribute

Whether to enable parallel execution of tool calls.

extra_headers: dict | str | None = None class-attribute instance-attribute

Additional headers to include in the request.

functions: list | str | None = None class-attribute instance-attribute

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

function_call: str | None = None class-attribute instance-attribute

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

base_url: str | None = environ.get('OPENAI_BASE_URL') class-attribute instance-attribute

Base URL for the API (default is None).

api_version: str | None = None class-attribute instance-attribute

API version (default is None).

api_key: str | None = None class-attribute instance-attribute

API key (default is None).

model_list: list | str | None = None class-attribute instance-attribute

List of api base, version, keys.

mock_response: str | None = None class-attribute instance-attribute

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

custom_llm_provider: str | None = 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: int | str | None = 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	LitellmParameters | ExpressionType[dict] | None	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: LitellmParameters | ExpressionType[dict] | None = 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: LitellmParameters | ExpressionType[dict] | None = 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	ExpressionType[dict[str, Any]] | None	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: ExpressionType[dict[str, Any]] | None = 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: ExpressionType[dict[str, Any]] | None = None class-attribute instance-attribute

Parameters sent to the model.

BaseCodeBlock

Bases: LeafBlock

Base class for code execution blocks.

Source code in src/pdl/pdl_ast.py

class BaseCodeBlock(LeafBlock):
    """Base class for code execution blocks."""

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

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.
scope	OptionalExpressionDictStr	Scope to use for the code execution. If not provided, the global scope is used.

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.
    """
    scope: OptionalExpressionDictStr = None
    """Scope to use for the code execution.  If not provided, the global scope is used.
    """

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.

scope: OptionalExpressionDictStr = None class-attribute instance-attribute

Scope to use for the code execution. If not provided, the global scope is used.

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.

JoinConfig

Bases: BaseModel

Configure how loop iterations or sequence of blocks should be combined.

Source code in src/pdl/pdl_ast.py

class JoinConfig(BaseModel):
    """Configure how loop iterations or sequence of blocks should be combined."""

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

JoinText

Bases: JoinConfig

Join loop iterations or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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 or sequence of blocks 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.

SequenceBlock

Bases: StructuredBlock

Generalization of the text, lastOf, array, and object blocks combining a list of blocks with a join operator.

Attributes:

Name	Type	Description
sequence	list[BlockType]	Sequence of blocks to join.
join	JoinType	Define how to combine the result of each block.

Source code in src/pdl/pdl_ast.py

class SequenceBlock(StructuredBlock):
    """Generalization of the `text`, `lastOf`, `array`, and `object` blocks combining a list of blocks with a `join` operator."""

    kind: Literal[BlockKind.SEQUENCE] = BlockKind.SEQUENCE
    sequence: list["BlockType"]
    """Sequence of blocks to join."""
    join: JoinType
    """Define how to combine the result of each block.
    """

sequence: list[BlockType] instance-attribute

Sequence of blocks to join.

join: JoinType instance-attribute

Define how to combine the result of each block.

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.

Attributes:

Name	Type	Description
object	dict[str, BlockType] | list[BlockType]	Object fields with their block definitions, or a list of blocks that produce objects to merge.

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"]
    """Object fields with their block definitions, or a list of blocks that produce objects to merge."""

object: dict[str, BlockType] | list[BlockType] instance-attribute

Object fields with their block definitions, or a list of blocks that produce objects to merge.

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.
tool_calls	OptionalExpressionList	List of tool invocations made by the assistant in this message.

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."""
    tool_calls: OptionalExpressionList = None
    """List of tool invocations made by the assistant in this message."""

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.

tool_calls: OptionalExpressionList = None class-attribute instance-attribute

List of tool invocations made by the assistant in this message.

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	PatternType | None	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: PatternType | None = 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: PatternType | None = 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.

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_	dict[str, ExpressionType[list]] | None	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_: dict[str, ExpressionType[list]] | None = 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: list["BlockType"] | None = None

for_: dict[str, ExpressionType[list]] | None = 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_	dict[str, ExpressionType[list]] | None	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_: dict[str, ExpressionType[list]] | None = 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: list["BlockType"] | None = None

for_: dict[str, ExpressionType[list]] | None = 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	str | None	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: str | None = None
    """Documentation associated to the aggregator config.
    """

description: str | None = 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[str | None]	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[str | None] = "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[str | None] = '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.

Attributes:

Name	Type	Description
aggregator	AggregatorType	Configuration for the aggregator (context or file-based).

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
    """Configuration for the aggregator (context or file-based)."""

aggregator: AggregatorType instance-attribute

Configuration for the aggregator (context or file-based).

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.

PdlBlock

Bases: RootModel

Wrapper class used to generate proper JSON Schema for PDL blocks.

This class introduces the BlockType in the generated JSON Schema, allowing for proper type definitions in schema documentation.

Source code in src/pdl/pdl_ast.py

class PdlBlock(RootModel):
    """Wrapper class used to generate proper JSON Schema for PDL blocks.

    This class introduces the BlockType in the generated JSON Schema,
    allowing for proper type definitions in schema documentation.
    """

    root: BlockType

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.
llm_usage	PdlUsage	Data structure where to accumulate LLMs usage.

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."""
    llm_usage: PdlUsage
    """Data structure where to accumulate LLMs usage."""

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.

llm_usage: PdlUsage instance-attribute

Data structure where to accumulate LLMs usage.

exec_program(prog: Program, config: InterpreterConfig | None = None, scope: ScopeType | Mapping[str, Any] | None = None, loc: PdlLocationType | None = 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	InterpreterConfig | None	Interpreter configuration. Defaults to None.	None
scope	ScopeType | Mapping[str, Any] | None	Environment defining the initial variables in scope to execute the program. Defaults to None.	None
loc	PdlLocationType | None	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: InterpreterConfig | None = None,
    scope: ScopeType | Mapping[str, Any] | None = None,
    loc: PdlLocationType | None = 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, ScopeType):
        scope = ScopeType(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":
            result_all: Result = {
                "result": result,
                "scope": future_scope.result(),
                "trace": trace,
                "replay": state.replay,
                "usage": state.llm_usage,
            }
            return result_all
        case _:
            assert False, 'The `output` variable should be "result" or "all"'

exec_dict(prog: dict[str, Any], config: InterpreterConfig | None = None, scope: ScopeType | Mapping[str, Any] | None = None, loc: PdlLocationType | None = 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	InterpreterConfig | None	Interpreter configuration. Defaults to None.	None
scope	ScopeType | Mapping[str, Any] | None	Environment defining the initial variables in scope to execute the program. Defaults to None.	None
loc	PdlLocationType | None	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: InterpreterConfig | None = None,
    scope: ScopeType | Mapping[str, Any] | None = None,
    loc: PdlLocationType | None = 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: InterpreterConfig | None = None, scope: ScopeType | Mapping[str, Any] | None = 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	InterpreterConfig | None	Interpreter configuration. Defaults to None.	None
scope	ScopeType | Mapping[str, Any] | None	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: InterpreterConfig | None = None,
    scope: ScopeType | Mapping[str, Any] | None = 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: InterpreterConfig | None = None, scope: ScopeType | Mapping[str, Any] | None = 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	InterpreterConfig | None	Interpreter configuration. Defaults to None.	None
scope	ScopeType | Mapping[str, Any] | None	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: InterpreterConfig | None = None,
    scope: ScopeType | Mapping[str, Any] | None = 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