chatomon.connectors.pydantic_ai ¶

Build an atom into a native PydanticAI agent.

Install with pip install 'chatomon[pydantic-ai]'.

Classes:

Name	Description
`PydanticAIConnector`	Builds an atom into a native `pydantic_ai.Agent`.

PydanticAIConnector ¶

PydanticAIConnector(
    model: Model | str,
    *,
    resolve_model: (
        Callable[[str], Model | str] | None
    ) = None,
    **agent_kwargs: Any
)

Bases: Connector['Agent[Any, str]']

Builds an atom into a native pydantic_ai.Agent.

chatomon's capability is the outermost layer; your own capabilities run inside it. Extra keyword arguments go to pydantic_ai.Agent unchanged (tools, deps_type, model_settings, ...). With output orbitals, use run() / run_sync(): streaming is refused because PydanticAI streams text before it can be checked. The reply must come from the model: a run in which a capability answers in its place or changes its reply fails with ChatomonConfigError.

Parameters:

Name	Type	Description	Default
`model`	`Model \| str`	The nucleus model: a PydanticAI model object or a `"provider:model"` string.	required
`resolve_model`	`Callable[[str], Model \| str] \| None`	Maps the model id of a JudgeOrbital to a PydanticAI model; by default the id is used as a model string.	`None`
`**agent_kwargs`	`Any`	Passed to `pydantic_ai.Agent`.	`{}`

Raises:

Type	Description
`ChatomonConfigError`	If `agent_kwargs` sets `instructions`, `system_prompt`, `output_type` or `name` (chatomon sets them from the atom).

Examples:

from pydantic_ai.models.test import TestModel

from chatomon import Atom
from chatomon.connectors.pydantic_ai import PydanticAIConnector

atom = Atom(
    name="speedypack",
    instructions="You are the customer assistant of SpeedyPack.",
    connector=PydanticAIConnector(TestModel(custom_output_text="Hi!")),
)
agent = atom.build()
assert agent.run_sync("hello").output == "Hi!"

Methods:

Name	Description
`build`	Build the native agent with chatomon's capability installed.
`generate_structured`	Make one structured call with a throwaway PydanticAI agent.

Source code in src/chatomon/connectors/pydantic_ai.py

def __init__(
    self,
    model: Model | str,
    *,
    resolve_model: Callable[[str], Model | str] | None = None,
    **agent_kwargs: Any,
) -> None:
    reserved = sorted(_RESERVED & set(agent_kwargs))
    if reserved:
        raise ChatomonConfigError(
            f"{', '.join(reserved)} is set by chatomon from the atom; "
            "remove it from PydanticAIConnector(...)"
        )
    self.model = model
    self.resolve_model = resolve_model
    self.agent_kwargs = agent_kwargs

build ¶

build(
    atom: Atom[Agent[Any, str]], engine: Engine
) -> Agent[Any, str]

Build the native agent with chatomon's capability installed.

Parameters:

Name	Type	Description	Default
`atom`	`Atom[Agent[Any, str]]`	The validated atom.	required
`engine`	`Engine`	The engine that runs the atom's orbitals.	required

Returns:

Type	Description
`Agent[Any, str]`	A `pydantic_ai.Agent` with text output.

Source code in src/chatomon/connectors/pydantic_ai.py

def build(self, atom: Atom[Agent[Any, str]], engine: Engine) -> Agent[Any, str]:
    """Build the native agent with chatomon's capability installed.

    Args:
        atom: The validated atom.
        engine: The engine that runs the atom's orbitals.

    Returns:
        A `pydantic_ai.Agent` with text output.
    """
    kwargs = dict(self.agent_kwargs)
    capabilities = [
        _ChatomonCapability(engine, atom.blocked_placeholder),
        *kwargs.pop("capabilities", ()),
    ]
    return Agent(
        self.model,
        name=atom.name,
        instructions=atom.system_prompt,
        capabilities=capabilities,
        **kwargs,
    )

generate_structured `async` ¶

generate_structured(
    model: str,
    *,
    system: str,
    prompt: str,
    schema: type[ModelT]
) -> ModelT

Make one structured call with a throwaway PydanticAI agent.

Parameters:

Name	Type	Description	Default
`model`	`str`	Model id; passed through `resolve_model` when set.	required
`system`	`str`	System instructions.	required
`prompt`	`str`	User prompt.	required
`schema`	`type[ModelT]`	Output type.	required

Returns:

Type	Description
`ModelT`	The validated output.

Source code in src/chatomon/connectors/pydantic_ai.py

async def generate_structured(
    self, model: str, *, system: str, prompt: str, schema: type[ModelT]
) -> ModelT:
    """Make one structured call with a throwaway PydanticAI agent.

    Args:
        model: Model id; passed through `resolve_model` when set.
        system: System instructions.
        prompt: User prompt.
        schema: Output type.

    Returns:
        The validated output.
    """
    resolved = self.resolve_model(model) if self.resolve_model is not None else model
    agent = Agent(resolved, instructions=system, output_type=schema)
    result = await agent.run(prompt)
    return result.output

chatomon.connectors.pydantic_ai ¶

PydanticAIConnector ¶

build ¶

generate_structured async ¶

generate_structured `async` ¶