Skip to content

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
LocalizedExpression

Expression with location information
Pattern
Block

Common fields for all PDL blocks.
FunctionBlock

Function declaration.
CallBlock

Calling a function.
LitellmParameters

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

Call a LLM through the LiteLLM API: https://docs.litellm.ai/.
CodeBlock

Execute a piece of code.
GetBlock

Get the value of a variable.
DataBlock

Arbitrary JSON value.
TextBlock

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

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

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

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

Create a message.
IfBlock

Conditional control structure.
MatchCase

Case of a match.
MatchBlock

Match control structure.
JoinConfig

Configure how loop iterations should be combined.
JoinText
JoinArray
JoinLastOf
ForBlock

Iteration over arrays.
RepeatUntilBlock

Repeat the execution of a block until a condition is satisfied.
ReadBlock

Read from a file or standard input.
IncludeBlock

Include a PDL file.
EmptyBlock

Block without an action. It can contain definitions.
Program

Prompt Declaration Language program (PDL)

Functions:

Name Description
get_default_model_parameters

Model-specific defaults to apply
get_sampling_defaults

Model-specific defaults to apply if we are sampling.

Attributes:

Name Type Description
AdvancedBlockType TypeAlias

Different types of structured blocks.
BlockType TypeAlias

All kinds of blocks.
BlockOrBlocksType TypeAlias

Block or list of blocks.

AdvancedBlockType: TypeAlias = FunctionBlock | CallBlock | LitellmModelBlock | CodeBlock | GetBlock | DataBlock | IfBlock | MatchBlock | RepeatUntilBlock | ForBlock | TextBlock | LastOfBlock | ArrayBlock | ObjectBlock | MessageBlock | ReadBlock | IncludeBlock | ErrorBlock | EmptyBlock module-attribute

Different types of structured blocks.

BlockType: TypeAlias = None | bool | int | float | str | AdvancedBlockType module-attribute

All kinds of blocks.

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

Block or list of blocks.

LocalizedExpression

Bases: BaseModel

Expression with location information

Source code in src/pdl/pdl_ast.py 
56
57
58
59
60
61
62
63
class LocalizedExpression(BaseModel):
    """Expression with location information"""

    model_config = ConfigDict(
        extra="forbid", use_attribute_docstrings=True, arbitrary_types_allowed=True
    )
    expr: Any
    location: Optional[LocationType] = None

Pattern

Bases: BaseModel

Attributes:

Name Type Description
assign Optional[str]

Name of the variable used to store the value matched by the pattern.
Source code in src/pdl/pdl_ast.py 
78
79
80
81
82
83
class Pattern(BaseModel):
    model_config = ConfigDict(extra="forbid")

    assign: Optional[str] = Field(default=None, alias="def")
    """Name of the variable used to store the value matched by the pattern.
    """

assign: Optional[str] = Field(default=None, alias='def') class-attribute instance-attribute

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

Block

Bases: BaseModel

Common fields for all PDL blocks.

Attributes:

Name Type Description
description Optional[str]

Documentation associated to the block.
spec Any

Type specification of the result of the block.
defs dict[str, BlockType]

Set of definitions executed before the execution of the block.
assign Optional[str]

Name of the variable used to store the result of the execution of the block.
contribute Sequence[ContributeTarget | dict[str, ContributeValue]]

Indicate if the block contributes to the result and background context.
parser Optional[ParserType]

Parser to use to construct a value out of a string result.
fallback Optional[BlockType]

Block to execute in case of error.
role RoleType

Role associated to the block and sub-blocks.
start_nanos Optional[int]

Time at which block execution began
end_nanos Optional[int]

Time at which block execution ended
timezone Optional[str]

Timezone of start_nanos and end_nanos
context Optional[ModelInput]

Current context
id Optional[str]

Unique identifier for this block
Source code in src/pdl/pdl_ast.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class Block(BaseModel):
    """Common fields for all PDL blocks."""

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

    description: Optional[str] = None
    """Documentation associated to the block.
    """
    spec: Any = None
    """Type specification of the result of the block.
    """
    defs: dict[str, "BlockType"] = {}
    """Set of definitions executed before the execution of the block.
    """
    assign: Optional[str] = Field(default=None, alias="def")
    """Name of the variable used to store the result of the execution of the block.
    """
    contribute: Sequence[ContributeTarget | dict[str, ContributeValue]] = [
        ContributeTarget.RESULT,
        ContributeTarget.CONTEXT,
    ]
    """Indicate if the block contributes to the result and background context.
    """
    parser: Optional[ParserType] = None
    """Parser to use to construct a value out of a string result."""
    fallback: Optional["BlockType"] = None
    """Block to execute in case of error.
    """
    role: RoleType = None
    """Role associated to the block and sub-blocks.
    """
    start_nanos: Optional[int] = 0
    """Time at which block execution began
    """
    end_nanos: Optional[int] = 0
    """Time at which block execution ended
    """
    timezone: Optional[str] = ""
    """Timezone of start_nanos and end_nanos
    """
    context: Optional[ModelInput] = []
    """Current context
    """
    id: Optional[str] = ""
    """Unique identifier for this block
    """
    # Fields for internal use
    result: Optional[Any] = None
    location: Optional[LocationType] = None

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

Documentation associated to the block.

spec: Any = 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.

assign: Optional[str] = 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[ContributeTarget | dict[str, ContributeValue]] = [ContributeTarget.RESULT, ContributeTarget.CONTEXT] class-attribute instance-attribute

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

parser: Optional[ParserType] = None class-attribute instance-attribute

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

fallback: Optional[BlockType] = None class-attribute instance-attribute

Block to execute in case of error.

role: RoleType = None class-attribute instance-attribute

Role associated to the block and sub-blocks.

start_nanos: Optional[int] = 0 class-attribute instance-attribute

Time at which block execution began

end_nanos: Optional[int] = 0 class-attribute instance-attribute

Time at which block execution ended

timezone: Optional[str] = '' class-attribute instance-attribute

Timezone of start_nanos and end_nanos

context: Optional[ModelInput] = [] class-attribute instance-attribute

Current context

id: Optional[str] = '' class-attribute instance-attribute

Unique identifier for this block

FunctionBlock

Bases: Block

Function declaration.

Attributes:

Name Type Description
function Optional[dict[str, Any]]

Functions parameters with their types.
returns BlockType

Body of the function
Source code in src/pdl/pdl_ast.py 
199
200
201
202
203
204
205
206
207
208
209
210
class FunctionBlock(Block):
    """Function declaration."""

    kind: Literal[BlockKind.FUNCTION] = BlockKind.FUNCTION
    function: Optional[dict[str, Any]]
    """Functions parameters with their types.
    """
    returns: "BlockType" = Field(..., alias="return")
    """Body of the function
    """
    # Field for internal use
    scope: SkipJsonSchema[Optional[ScopeType]] = Field(default=None, repr=False)

function: Optional[dict[str, Any]] instance-attribute

Functions parameters with their types.

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

Body of the function

CallBlock

Bases: Block

Calling a function.

Attributes:

Name Type Description
call ExpressionType

Function to call.
args ExpressionType

Arguments of the function with their values.
Source code in src/pdl/pdl_ast.py 
213
214
215
216
217
218
219
220
221
222
223
224
class CallBlock(Block):
    """Calling a function."""

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

call: ExpressionType instance-attribute

Function to call.

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

Arguments of the function with their values.

LitellmParameters

Bases: BaseModel

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

Attributes:

Name Type Description
timeout Optional[Union[float, str]] | str

Timeout in seconds for completion requests (Defaults to 600 seconds).
temperature Optional[float] | str

The temperature parameter for controlling the randomness of the output (default is 1.0).
top_p Optional[float] | str

The top-p parameter for nucleus sampling (default is 1.0).
n Optional[int] | str

The number of completions to generate (default is 1).
stop Optional[str | list[str]] | str

Up to 4 sequences where the LLM API will stop generating further tokens.
max_tokens Optional[int] | str

The maximum number of tokens in the generated completion (default is infinity).
presence_penalty Optional[float] | str

It is used to penalize new tokens based on their existence in the text so far.
frequency_penalty Optional[float] | str

It is used to penalize new tokens based on their frequency in the text so far.
logit_bias Optional[dict] | str

Used to modify the probability of specific tokens appearing in the completion.
user Optional[str] | str

A unique identifier representing your end-user. This can help the LLM provider to monitor and detect abuse.
logprobs Optional[bool] | str

Whether to return log probabilities of the output tokens or not. If true, returns the log probabilities of each output token returned in the content of message
top_logprobs Optional[int] | str

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

Additional headers to include in the request.
functions Optional[list] | str

A list of functions to apply to the conversation messages (default is an empty list)
function_call Optional[str] | str

The name of the function to call within the conversation (default is an empty string)
base_url Optional[str] | str

Base URL for the API (default is None).
api_version Optional[str] | str

API version (default is None).
api_key Optional[str] | str

API key (default is None).
model_list Optional[list] | str

List of api base, version, keys.
mock_response Optional[str] | str

If provided, return a mock completion response for testing or debugging purposes (default is None).
custom_llm_provider Optional[str] | str

Used for Non-OpenAI LLMs, Example usage for bedrock, set model="amazon.titan-tg1-large" and custom_llm_provider="bedrock"
max_retries Optional[int] | str

The number of retries to attempt (default is 0).
Source code in src/pdl/pdl_ast.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
class LitellmParameters(BaseModel):
    """Parameters passed to LiteLLM. More details at https://docs.litellm.ai/docs/completion/input."""

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Additional headers to include in the request.

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

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

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

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

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

Base URL for the API (default is None).

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

API version (default is None).

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

API key (default is None).

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

List of api base, version, keys.

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

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

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

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

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

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

LitellmModelBlock

Bases: ModelBlock

Call a LLM through the LiteLLM API: https://docs.litellm.ai/.

Source code in src/pdl/pdl_ast.py 
327
328
329
330
331
class LitellmModelBlock(ModelBlock):
    """Call a LLM through the LiteLLM API: https://docs.litellm.ai/."""

    platform: Literal[ModelPlatform.LITELLM] = ModelPlatform.LITELLM
    parameters: Optional[LitellmParameters | ExpressionType] = None

CodeBlock

Bases: Block

Execute a piece of code.

Attributes:

Name Type Description
lang Literal['python', 'command', 'jinja', 'pdl']

Programming language of the code.
code BlockType

Code to execute.
Source code in src/pdl/pdl_ast.py 
334
335
336
337
338
339
340
341
342
343
class CodeBlock(Block):
    """Execute a piece of code."""

    kind: Literal[BlockKind.CODE] = BlockKind.CODE
    lang: Literal["python", "command", "jinja", "pdl"]
    """Programming language of the code.
    """
    code: "BlockType"
    """Code to execute.
    """

lang: Literal['python', 'command', 'jinja', 'pdl'] instance-attribute

Programming language of the code.

code: BlockType instance-attribute

Code to execute.

GetBlock

Bases: Block

Get the value of a variable.

Attributes:

Name Type Description
get str

Name of the variable to access.
Source code in src/pdl/pdl_ast.py 
346
347
348
349
350
351
class GetBlock(Block):
    """Get the value of a variable."""

    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: Block

Arbitrary JSON value.

Attributes:

Name Type Description
data ExpressionType

Value defined.
raw bool

Do not evaluate expressions inside strings.
Source code in src/pdl/pdl_ast.py 
354
355
356
357
358
359
360
361
class DataBlock(Block):
    """Arbitrary JSON value."""

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

data: ExpressionType instance-attribute

Value defined.

raw: bool = False class-attribute instance-attribute

Do not evaluate expressions inside strings.

TextBlock

Bases: Block

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 
364
365
366
367
368
369
370
class TextBlock(Block):
    """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: Block

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

Source code in src/pdl/pdl_ast.py 
373
374
375
376
377
class LastOfBlock(Block):
    """Return the value of the last block if the list of blocks."""

    kind: Literal[BlockKind.LASTOF] = BlockKind.LASTOF
    lastOf: list["BlockType"]

ArrayBlock

Bases: Block

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

Source code in src/pdl/pdl_ast.py 
380
381
382
383
384
class ArrayBlock(Block):
    """Return the array of values computed by each block of the list of blocks."""

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

ObjectBlock

Bases: Block

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

Source code in src/pdl/pdl_ast.py 
387
388
389
390
391
class ObjectBlock(Block):
    """Return the object where the value of each field is defined by a block. If the body of the object is an array, the resulting object is the union of the objects computed by each element of the array."""

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

MessageBlock

Bases: Block

Create a message.

Attributes:

Name Type Description
role RoleType

Role of associated to the message.
content BlockType

Content of the message.
Source code in src/pdl/pdl_ast.py 
394
395
396
397
398
399
400
401
class MessageBlock(Block):
    """Create a message."""

    kind: Literal[BlockKind.MESSAGE] = BlockKind.MESSAGE
    role: RoleType  # pyright: ignore
    """Role of associated to the message."""  # pyright: ignore
    content: "BlockType"
    """Content of the message."""

role: RoleType instance-attribute

Role of associated to the message.

content: BlockType instance-attribute

Content of the message.

IfBlock

Bases: Block

Conditional control structure.

Attributes:

Name Type Description
condition ExpressionType

Condition.
then BlockType

Branch to exectute if the condition is true.
elses Optional[BlockType]

Branch to execute if the condition is false.
Source code in src/pdl/pdl_ast.py 
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
class IfBlock(Block):
    """Conditional control structure."""

    kind: Literal[BlockKind.IF] = BlockKind.IF
    condition: ExpressionType = Field(alias="if")
    """Condition.
    """
    then: "BlockType"
    """Branch to exectute if the condition is true.
    """
    elses: Optional["BlockType"] = Field(default=None, alias="else")
    """Branch to execute if the condition is false.
    """
    # Field for internal use
    if_result: Optional[bool] = None

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

Condition.

then: BlockType instance-attribute

Branch to exectute if the condition is true.

elses: Optional[BlockType] = Field(default=None, alias='else') class-attribute instance-attribute

Branch to execute if the condition is false.

MatchCase

Bases: BaseModel

Case of a match.

Attributes:

Name Type Description
case Optional[PatternType]

Value to match.
if_ Optional[ExpressionType]

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 
421
422
423
424
425
426
427
428
429
430
431
432
433
class MatchCase(BaseModel):
    """Case of a match."""

    model_config = ConfigDict(extra="forbid")
    case: Optional[PatternType] = None
    """Value to match.
    """
    if_: Optional[ExpressionType] = Field(default=None, alias="if")
    """Boolean condition to satisfy.
    """
    then: "BlockType"
    """Branch to execute if the value is matched and the condition is satisfied.
    """

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

Value to match.

if_: Optional[ExpressionType] = 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: Block

Match control structure.

Attributes:

Name Type Description
match_ ExpressionType

Matched expression.
Source code in src/pdl/pdl_ast.py 
436
437
438
439
440
441
442
443
class MatchBlock(Block):
    """Match control structure."""

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

match_: ExpressionType = Field(alias='match') class-attribute instance-attribute

Matched expression.

JoinConfig

Bases: BaseModel

Configure how loop iterations should be combined.

Source code in src/pdl/pdl_ast.py 
452
453
454
455
class JoinConfig(BaseModel):
    """Configure how loop iterations should be combined."""

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

JoinText

Bases: JoinConfig

Attributes:

Name Type Description
iteration_type Literal[TEXT]

String concatenation of the result of each iteration.
join_string str

String used to concatenate each iteration of the loop.
Source code in src/pdl/pdl_ast.py 
458
459
460
461
462
463
464
465
466
467
class JoinText(JoinConfig):
    iteration_type: Literal[IterationType.TEXT] = Field(
        alias="as", default=IterationType.TEXT
    )
    """String concatenation of the result of each iteration.
    """

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

iteration_type: Literal[IterationType.TEXT] = Field(alias='as', default=IterationType.TEXT) class-attribute instance-attribute

String concatenation of the result of each iteration.

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

String used to concatenate each iteration of the loop.

JoinArray

Bases: JoinConfig

Attributes:

Name Type Description
iteration_type Literal[ARRAY]

Return the result of each iteration as an array.
Source code in src/pdl/pdl_ast.py 
470
471
472
473
class JoinArray(JoinConfig):
    iteration_type: Literal[IterationType.ARRAY] = Field(alias="as")
    """Return the result of each iteration as an array.
    """

iteration_type: Literal[IterationType.ARRAY] = Field(alias='as') class-attribute instance-attribute

Return the result of each iteration as an array.

JoinLastOf

Bases: JoinConfig

Attributes:

Name Type Description
iteration_type Literal[LASTOF]

Return the result of the last iteration.
Source code in src/pdl/pdl_ast.py 
476
477
478
479
class JoinLastOf(JoinConfig):
    iteration_type: Literal[IterationType.LASTOF] = Field(alias="as")
    """Return the result of the last iteration.
    """

iteration_type: Literal[IterationType.LASTOF] = Field(alias='as') class-attribute instance-attribute

Return the result of the last iteration.

ForBlock

Bases: Block

Iteration over arrays.

Attributes:

Name Type Description
fors dict[str, ExpressionType]

Arrays to iterate over.
repeat BlockType

Body of the loop.
join JoinType

Define how to combine the result of each iteration.
Source code in src/pdl/pdl_ast.py 
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
class ForBlock(Block):
    """Iteration over arrays."""

    kind: Literal[BlockKind.FOR] = BlockKind.FOR
    fors: dict[str, ExpressionType] = Field(alias="for")
    """Arrays to iterate over.
    """
    repeat: "BlockType"
    """Body of the loop.
    """
    join: JoinType = JoinText()
    """Define how to combine the result of each iteration.
    """
    # Field for internal use
    trace: Optional[list["BlockType"]] = None

fors: dict[str, ExpressionType] = Field(alias='for') class-attribute instance-attribute

Arrays to iterate over.

repeat: BlockType instance-attribute

Body of the loop.

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

Define how to combine the result of each iteration.

RepeatUntilBlock

Bases: Block

Repeat the execution of a block until a condition is satisfied.

Attributes:

Name Type Description
repeat BlockType

Body of the loop.
until ExpressionType

Condition to exit at the end of the loop.
max_iterations Optional[ExpressionType]

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 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
class RepeatUntilBlock(Block):
    """Repeat the execution of a block until a condition is satisfied."""

    kind: Literal[BlockKind.REPEAT_UNTIL] = BlockKind.REPEAT_UNTIL
    repeat: "BlockType"
    """Body of the loop.
    """
    until: ExpressionType = False
    """Condition to exit at the end of the loop.
    """
    max_iterations: Optional[ExpressionType] = None
    """Maximal number of iterations to perform.
    """
    join: JoinType = JoinText()
    """Define how to combine the result of each iteration.
    """
    # Field for internal use
    trace: Optional[list["BlockType"]] = None

repeat: BlockType instance-attribute

Body of the loop.

until: ExpressionType = False class-attribute instance-attribute

Condition to exit at the end of the loop.

max_iterations: Optional[ExpressionType] = 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.

ReadBlock

Bases: Block

Read from a file or standard input.

Attributes:

Name Type Description
read ExpressionType | None

Name of the file to read. If None, read the standard input.
message Optional[str]

Message to prompt the user to enter a value.
multiline bool

Indicate if one or multiple lines shoud be read.
Source code in src/pdl/pdl_ast.py 
522
523
524
525
526
527
528
529
530
531
532
533
534
class ReadBlock(Block):
    """Read from a file or standard input."""

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

read: ExpressionType | None instance-attribute

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

message: Optional[str] = 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 shoud be read.

IncludeBlock

Bases: Block

Include a PDL file.

Attributes:

Name Type Description
include str

Name of the file to include.
Source code in src/pdl/pdl_ast.py 
537
538
539
540
541
542
543
544
545
class IncludeBlock(Block):
    """Include a PDL file."""

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

include: str instance-attribute

Name of the file to include.

EmptyBlock

Bases: Block

Block without an action. It can contain definitions.

Source code in src/pdl/pdl_ast.py 
554
555
556
557
class EmptyBlock(Block):
    """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 
591
592
593
594
595
596
597
598
class Program(RootModel):
    """
    Prompt Declaration Language program (PDL)
    """

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

root: BlockType instance-attribute

Entry point to parse a PDL program using Pydantic.

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

Model-specific defaults to apply

Source code in src/pdl/pdl_ast.py 
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
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
        {
            "replicate*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|>",
                    },
                    "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|>",
            }
        },
    ]

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

Model-specific defaults to apply if we are sampling.

Source code in src/pdl/pdl_ast.py 
756
757
758
759
760
761
762
763
764
765
766
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.
Source code in src/pdl/pdl.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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."""

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.

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

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

Parameters:

Name Type Description Default
prog Program

Program to execute.

 required
config Optional[InterpreterConfig]

Interpreter configuration. Defaults to None.

 None
scope Optional[ScopeType | dict[str, Any]]

Environment defining the initial variables in scope to execute the program. Defaults to None.

 None
loc Optional[LocationType]

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, and trace.
Source code in src/pdl/pdl.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def exec_program(
    prog: Program,
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    loc: Optional[LocationType] = 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`, and `trace`.
    """
    logging.basicConfig(filename="log.txt", encoding="utf-8", format="", filemode="w")
    config = config or {}
    state = InterpreterState(**config)
    if not isinstance(scope, PdlDict):
        scope = PdlDict(scope or {})
    loc = loc or empty_block_location
    future_result, _, future_scope, trace = process_prog(state, scope, prog, loc)
    result = future_result.result()
    match output:
        case "result":
            return result
        case "all":
            scope = future_scope.result()
            return {"result": result, "scope": scope, "trace": trace}
        case _:
            assert False, 'The `output` variable should be "result" or "all"'

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

Execute a PDL program given as a dictionary.

Parameters:

Name Type Description Default
prog dict[str, Any]

Program to execute.

 required
config Optional[InterpreterConfig]

Interpreter configuration. Defaults to None.

 None
scope Optional[ScopeType | dict[str, Any]]

Environment defining the initial variables in scope to execute the program. Defaults to None.

 None
loc Optional[LocationType]

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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def exec_dict(
    prog: dict[str, Any],
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    loc: Optional[LocationType] = 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 = Program.model_validate(prog)
    result = exec_program(program, config, scope, loc, output)
    return result

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

Execute a PDL program given as YAML string.

Parameters:

Name Type Description Default
prog str

Program to execute.

 required
config Optional[InterpreterConfig]

Interpreter configuration. Defaults to None.

 None
scope Optional[ScopeType | dict[str, Any]]

Environment defining the initial variables in scope to execute the program. Defaults to None.

 None
output Literal['result', 'all']

Configure the output of the returned value of this function. Defaults to "result"

 'result'

Returns:

Type Description
Any

Return the final result.
Source code in src/pdl/pdl.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def exec_str(
    prog: str,
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    output: Literal["result", "all"] = "result",
) -> Any:
    """Execute a PDL program given as YAML string.

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

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

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

Execute a PDL program given as YAML file.

Parameters:

Name Type Description Default
prog str | Path

Program to execute.

 required
config Optional[InterpreterConfig]

Interpreter configuration. Defaults to None.

 None
scope Optional[ScopeType | dict[str, Any]]

Environment defining the initial variables in scope to execute the program. Defaults to None.

 None
output Literal['result', 'all']

Configure the output of the returned value of this function. Defaults to "result"

 'result'

Returns:

Type Description
Any

Return the final result.
Source code in src/pdl/pdl.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def exec_file(
    prog: str | Path,
    config: Optional[InterpreterConfig] = None,
    scope: Optional[ScopeType | dict[str, Any]] = None,
    output: Literal["result", "all"] = "result",
) -> Any:
    """Execute a PDL program given as YAML file.

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

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