reflexr.agent¶

Agent and graph actions: pydantic-ai and pydantic-graph as adapters of the action port.

An AgentAction runs a pydantic-ai agent whose deps_type is Reaction; the EventContext capability gives it tools to read the log and emit events. A GraphAction runs a pydantic-graph graph, checkpointed at its step boundaries so a retry resumes after the last completed step. function_model scripts a model for tests.

Agents¶

pydantic-ai agents as actions. See Agents.

AgentAction `dataclass` ¶

AgentAction(
    agent: Agent[Reaction[D], O],
    *,
    name: str = "",
    prompt: str
    | Callable[[Reaction[D]], str]
    | None = None,
    usage_limits: UsageLimits | None = None,
    params: type[BaseModel] | None = None,
)

Run a pydantic-ai agent in response to a firing.

The agent's deps_type is Reaction[D]; add the EventContext capability to give it tools for the log. By default the prompt describes the firing: the rule, the scope and the events that made it fire. The agent's output is the run's output. Each run joins its causal chain's conversation, so the traces of one incident are one session.

Parameters:

Name	Type	Description	Default
`agent`	`Agent[Reaction[D], O]`	The agent.	required
`name`	`str`	The name rules refer to the action by; defaults to the agent's name.	`''`
`prompt`	`str \| Callable[[Reaction[D]], str] \| None`	The user prompt, or a function of the reaction that returns it. Defaults to a description of the firing.	`None`
`usage_limits`	`UsageLimits \| None`	Limits on requests and tokens per attempt.	`None`
`params`	`type[BaseModel] \| None`	The model of the params rules pass the action. Its prompt and tools read them with `reaction.params_as(Model)`, naming the model again.	`None`

EventContext `dataclass` ¶

EventContext(
    emit: Sequence[type[Event]] = (),
    *,
    max_event_chars: int = 2000,
    read_limit: int = 50,
    max_retries: int = 3,
)

Bases: AbstractCapability[Reaction[Any]]

Make an agent a response to a firing.

Parameters:

Name	Type	Description	Default
`emit`	`Sequence[type[Event]]`	The event types the agent may publish with the `emit_event` tool. Empty, the default, gives it no way to publish.	`()`
`max_event_chars`	`int`	How much of each event `read_events` shows the model.	`2000`
`read_limit`	`int`	The most envelopes one `read_events` call returns.	`50`
`max_retries`	`int`	How many times the model may retry a refused tool call.	`3`

get_toolset ¶

get_toolset() -> FunctionToolset[Reaction[Any]]

Return the tools for reading the log and emitting events.

get_instructions ¶

get_instructions() -> str

Return the instructions: how to respond, and what the agent may emit.

wrap_run `async` ¶

wrap_run(
    ctx: Context, *, handler: WrapRunHandler
) -> AgentRunResult[Any]

Attribute the agent's span to its workspace, rule, run and causal chain.

Graphs¶

pydantic-graph graphs as actions, checkpointed at their step boundaries. See Graphs.

GraphAction `dataclass` ¶

GraphAction(
    graph: Graph[S, Reaction[D], I, O],
    *,
    name: str = "",
    state: Callable[[Reaction[D]], S] | None = None,
    inputs: Callable[[Reaction[D]], I] | None = None,
    input_types: Mapping[str, Any] = dict[str, Any](),
    params: type[BaseModel] | None = None,
)

Run a pydantic-graph graph in response to a firing, checkpointed after its steps.

The graph's deps are the Reaction, so steps can read the firing and emit events. Its output is the run's output.

A checkpoint saves the next node's inputs as that node's input type. A step's is its StepContext annotation, and the end node's the graph's output type. A fork's is inferred from the edges into it: the return type of the steps they come from, or the graph's input type from its start. It stays unknown when an edge has a transform, or comes from a step with no return annotation (or Any), a decision, a join or a fork, or when the edges carry different types. A boundary before a node whose input type is unknown is not saved, and neither is one before a decision, which runs no code: the boundary after it saves the same.

Nor is a boundary whose state, graph inputs or next inputs would not read back equal to what they were, such as a model given to a stream step or a BaseNode, whose input types say Any.

Parameters:

Name	Type	Description	Default
`graph`	`Graph[S, Reaction[D], I, O]`	A graph built with `GraphBuilder`, whose deps type is `Reaction[D]`.	required
`name`	`str`	The name rules refer to the action by; defaults to the graph's name.	`''`
`state`	`Callable[[Reaction[D]], S] \| None`	Builds the graph's initial state from the reaction; defaults to the state type's constructor with no arguments.	`None`
`inputs`	`Callable[[Reaction[D]], I] \| None`	Builds the graph's inputs from the reaction; defaults to None.	`None`
`input_types`	`Mapping[str, Any]`	Input types by node id, for the steps and forks whose type reflexr cannot read or infer, such as a stream step, whose input type reads as `Any`, or a fork after a transform. An explicit type wins over an annotation or an inferred one.	`dict[str, Any]()`
`params`	`type[BaseModel] \| None`	The model of the params rules pass the action. Its steps, and `state` and `inputs`, read them with `reaction.params_as(Model)`, naming the model again.	`None`

Raises:

Type	Description
`ValueError`	If the action has no name, or `input_types` names a node that is not a step or a fork.

Scripting a model¶

A model for tests, answering plain and streamed requests from one function. See Scripting agents.

function_model ¶

function_model(respond: Respond) -> FunctionModel

Return a pydantic-ai FunctionModel that answers every request with respond.

A plain request gets the response as it is. A streamed request, such as each request of an agent the LiteLLMGateway capability wraps, gets it streamed: each text part as one text delta, each thinking part as one thinking delta, and each tool call whole. Script a model for a test with it, rather than writing a stream function beside the function::

def respond(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    return ModelResponse(parts=[TextPart("On it.")])


with agent.override(model=function_model(respond)):
    ...

Parameters:

Name	Type	Description	Default
`respond`	`Respond`	Answers each request from the messages so far, as `FunctionModel`'s function does; it may be async.	required

Raises:

Type	Description
`ValueError`	When a streamed response has a part other than text, thinking or a tool call, which the stream cannot carry.

Respond ¶

Respond = Callable[
    [list[ModelMessage], AgentInfo],
    ModelResponse | Awaitable[ModelResponse],
]

Answers a model request: pydantic-ai's FunctionDef, sync or async.

Describing a firing¶

How firings and envelopes are shown to a model. Use them in your own prompts.

firing_text ¶

firing_text(
    reaction: Reaction[Any], *, max_event_chars: int = 2000
) -> str

Describe the firing an action is responding to: the rule, the scope and the events.

envelope_text ¶

envelope_text(
    envelope: Envelope, *, max_chars: int = 2000
) -> str

Render one envelope as a line a model can read.

reflexr.agent¶

Agents¶

AgentAction dataclass ¶

EventContext dataclass ¶

get_toolset ¶

get_instructions ¶

wrap_run async ¶

Graphs¶

GraphAction dataclass ¶

Scripting a model¶

function_model ¶

Respond ¶

Describing a firing¶

firing_text ¶

envelope_text ¶

AgentAction `dataclass` ¶

EventContext `dataclass` ¶

wrap_run `async` ¶

GraphAction `dataclass` ¶