> ## Documentation Index
> Fetch the complete documentation index at: https://launchpad.datalumina.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Agent Node

> Integrate LLMs into workflows with AgentNode, backed by pydantic-ai.

The `AgentNode` class wraps a pydantic-ai `Agent` so LLM calls slot into the same Chain-of-Responsibility pattern as any other node. Subclasses implement two methods:

* **`get_agent_config()`** — returns an `AgentConfig` describing the model provider, model name, structured output type, tools, and instructions.
* **`process()`** — runs the underlying `Agent`, validates outputs against `OutputType`, and saves them with `save_output()`.

See [LLM Providers](/docs/tools/llm-providers) for per-provider configuration details.

## AgentConfig

```python theme={null}
@dataclass
class AgentConfig:
    model_provider: ModelProvider
    model_name: Union[
        OpenAIModelName, AnthropicModelName, GeminiModelName, BedrockModelName
    ]
    output_type: Any = str
    instructions: Optional[str] = None
    deps_type: Optional[Type[Any]] = None
    name: str | None = None
    model_settings: ModelSettings | None = None
    retries: int = 1
    output_retries: int | None = None
    tools: List = field(default_factory=list)
    builtin_tools: List = field(default_factory=list)
    instrument: bool = True
```

Notable fields:

* **`instructions`** — the system prompt for the underlying pydantic-ai `Agent`. You can also register per-run context via the `@self.agent.instructions` decorator inside `process()`.
* **`output_type`** — a Pydantic model (subclass of `AgentNode.OutputType`) for structured output, or `str` for plain text.
* **`deps_type`** — a Pydantic model exposed via `RunContext` so tools and instruction callbacks can read dependencies the node computes at runtime.
* **`tools`, `builtin_tools`** — pydantic-ai tool definitions.
* **`instrument`** — defaults to `True`; set `False` to opt this node out of Langfuse instrumentation even when the workflow is traced.

## AgentNode base class

```python theme={null}
class AgentNode(Node, ABC):
    class DepsType(BaseModel):
        pass

    class OutputType(BaseModel):
        pass

    def __init__(self, task_context: TaskContext = None):
        super().__init__(task_context=task_context)
        self.__async_client = AsyncClient()
        agent_wrapper = self.get_agent_config()
        self.agent = Agent(
            model=self.__get_model_instance(
                agent_wrapper.model_provider, agent_wrapper.model_name
            ),
            output_type=agent_wrapper.output_type,
            instructions=agent_wrapper.instructions,
            deps_type=agent_wrapper.deps_type,
            name=agent_wrapper.name,
            model_settings=agent_wrapper.model_settings,
            retries=agent_wrapper.retries,
            output_retries=agent_wrapper.output_retries,
            tools=agent_wrapper.tools,
            builtin_tools=agent_wrapper.builtin_tools,
            instrument=agent_wrapper.instrument,
        )
        self.agent.instrument_all()

    @abstractmethod
    def get_agent_config(self) -> AgentConfig:
        pass

    @abstractmethod
    async def process(self, task_context: TaskContext) -> TaskContext:
        pass
```

<Warning>
  `AgentNode` runs asynchronously. Call `await self.agent.run(...)` or `async with self.agent.run_stream(...)` — pydantic-ai's `run_sync()` will not work inside the workflow's event loop.
</Warning>

## Implementation examples

### Without dependencies

```python theme={null}
class FilterSpamNode(AgentNode):
    class OutputType(AgentNode.OutputType):
        reasoning: str = Field(description="Reasoning for the spam classification.")
        confidence: float = Field(ge=0, le=1)
        is_human: bool = Field(description="True if the message is from a human.")

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            instructions=(
                "You are a helpful assistant that filters messages to determine "
                "whether they are written by a human or are spam generated by a bot."
            ),
            output_type=self.OutputType,
            model_provider=ModelProvider.OPENAI,
            model_name="gpt-5.4-mini",
        )

    async def process(self, task_context: TaskContext) -> TaskContext:
        event: CustomerCareEventSchema = task_context.event
        result = await self.agent.run(user_prompt=event.model_dump_json())
        self.save_output(result.output)
        return task_context
```

### With dependencies and dynamic instructions

Use `deps_type` + `@self.agent.instructions` to inject runtime context (for example, retrieval results in a RAG node) into the system prompt at call time:

```python theme={null}
class GenerationNode(AgentNode):
    class DepsType(AgentNode.DepsType):
        context: RetrievalResults

    class OutputType(AgentNode.OutputType):
        answer: str
        sources: list[str]
        confidence: float = Field(ge=0, le=1)

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            instructions=(
                "You are a helpful assistant that answers questions using the "
                "retrieved documents."
            ),
            output_type=GenerationNode.OutputType,
            deps_type=GenerationNode.DepsType,
            model_provider=ModelProvider.OPENAI,
            model_name="gpt-5.4-mini",
        )

    async def process(self, task_context: TaskContext) -> TaskContext:
        retrieval: RetrievalNode.OutputType = self.get_output(RetrievalNode)
        deps = GenerationNode.DepsType(context=retrieval.results)

        @self.agent.instructions
        def add_rag_context(ctx: RunContext[GenerationNode.DepsType]) -> str:
            return (
                "Here are the documents I found for your query:\n"
                f"{ctx.deps.context.model_dump_json(indent=2)}"
            )

        result = await self.agent.run(
            user_prompt=task_context.event.query,
            deps=deps,
        )
        self.save_output(result.output)
        return task_context
```

## Key features

* **Type-safe outputs** — Pydantic `OutputType` validates what the model returns.
* **Flexible dependencies** — `DepsType` + `RunContext` plug structured context into tools and instructions.
* **Seven model providers** — switch between OpenAI, Azure OpenAI, Anthropic, Google Gemini, Google Vertex AI, Bedrock, and Ollama by changing one enum.
* **Dynamic instructions** — augment the system prompt per run via `@self.agent.instructions`.
* **Tool integration** — register pydantic-ai tools or built-in tools via `AgentConfig.tools` / `builtin_tools`.
* **Langfuse-ready** — tracing is on by default via `instrument=True`; the workflow decides whether traces are actually exported.