Agent Loop — let the LLM call your tools¶

→ Documentation index

extract_data_with_tools and run_extractor_agent extend structured extraction with an agentic tool loop: the LLM may call caller-supplied tools any number of times — look up data, create resources, trigger side effects — and then delivers a final answer that is validated against your Pydantic schema.

Use it when the answer cannot be produced from the prompt text alone:

the model needs data that only your application can provide (database rows, API responses, file contents),
a resource has to be created first and its id must appear in the final answer,
the extraction requires a decision based on intermediate lookups.

If the model only needs to read the text you already have, use extract_data_from_text / extract_data instead — they are simpler and cheaper.

Quick example¶

import asyncio
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from saidex import Tool, extract_data_with_tools

# 1. Describe the tool's arguments with a Pydantic model
class OrderStatusArgs(BaseModel):
    order_id: str = Field(description="Order id, e.g. ORD-1042")

# 2. Provide an async handler — its return value is shown to the LLM
async def order_status_handler(order_id: str) -> dict:
    # ... real lookup ...
    return {"order_id": order_id, "status": "shipped", "eta": "2026-06-15"}

order_status = Tool(
    name="get_order_status",
    description="Look up the current status and ETA of an order.",
    parameters=OrderStatusArgs,
    handler=order_status_handler,
)

# 3. Define the final-answer schema and run the loop
class TicketResolution(BaseModel):
    order_id: str
    current_status: str
    customer_reply: str = Field(description="Friendly reply for the customer")

async def main() -> None:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    resolution, stats = await extract_data_with_tools(
        llm,
        TicketResolution,
        "Customer asks: where is my order ORD-1042?",
        tools=[order_status],
    )

    print(resolution)
    print(f"iterations={stats.iterations} tool_calls={stats.tool_calls}")

asyncio.run(main())

Full runnable example: examples/08_agent_tools.py

How the loop works¶

┌────────────────────────────────────────────────────────┐
│  LLM  (your tools + a final-answer tool are bound)     │
└───────────────┬────────────────────────────────────────┘
                │
   ┌────────────┴─────────────┐
   │ calls a helper tool?      │──── yes ──► validate args → run handler
   │                           │            append result as ToolMessage
   │                           │            └──► back to the LLM
   │ calls the final-answer    │
   │ tool (or emits JSON)?     │──── yes ──► validate against schema
   └───────────────────────────┘             ├─ valid   → return (instance, stats)
                                             └─ invalid → feed error report back,
                                                          continue the loop

Every Tool is converted to an OpenAI-style function definition and bound to the model, together with a final-answer tool generated from schema.
When the LLM calls a helper tool, the arguments are validated against the tool's parameters model, the handler runs, and the result is appended to the conversation as a ToolMessage.
When the LLM submits the final answer, it is validated with Pydantic. On failure, the field-level error report is sent back and the loop continues.
The loop ends with a validated instance — or None once max_iterations is reached or a final answer is rejected with no validation retries left (max_validation_retries), after optionally retrying on a fallback model.

Two failure-safety properties worth knowing:

Handler exceptions never crash the loop. They are caught and returned to the LLM as the tool result ("Tool 'x' raised an error: …"), so the model can retry or work around the failure.
Invalid tool arguments are reported, not raised. The LLM receives a structured validation message and can correct its call.

The `Tool` dataclass¶

from saidex import Tool

Field	Type	Description
`name`	`str`	Function name exposed to the LLM (valid identifier)
`description`	`str`	What the tool does — the LLM decides from this when to call it
`parameters`	`type[BaseModel]`	Pydantic model whose fields define the arguments; the JSON Schema is generated automatically
`handler`	`Callable[..., Awaitable[Any]]`	Async callable receiving the validated arguments as keyword arguments

The handler's return value becomes the ToolMessage content: strings are passed through, everything else is serialised with json.dumps (falling back to str). Return compact, informative results — the LLM has to read them.

class SearchArgs(BaseModel):
    query: str = Field(description="Search term")
    limit: int = Field(default=5, ge=1, le=20, description="Max results")

async def search_handler(query: str, limit: int = 5) -> list[dict]:
    return await my_database.search(query, limit=limit)

search = Tool(
    name="search_products",
    description="Search the product catalogue by free-text query.",
    parameters=SearchArgs,
    handler=search_handler,
)

Tips for reliable tools:

Write description and field descriptions as carefully as your extraction schema — they are the only things the LLM sees.
Keep results small and structured; trim large payloads to the fields the model actually needs.
Tools may be called multiple times and in any order — make handlers idempotent where possible, or say in the description when a tool must be called only once.

`extract_data_with_tools` parameters¶

Parameter	Type	Default	Description
`llm_model`	`Any`	—	LangChain chat model with `bind_tools` support
`schema`	`type[ModelT]`	—	Pydantic model for the final answer
`text`	`str`	—	Task description (becomes the `HumanMessage`)
`tools`	`list[Tool]`	—	Helper tools the LLM may call freely
`final_answer_mode`	`ExtractionMode`	`TOOL_CALLING`	How the final answer is collected (see below)
`system_prompt`	`str \\| None`	auto	System instruction (a generic one is used if omitted)
`callbacks`	`list[Any] \\| None`	`None`	LangChain callback handlers
`fallback_llm_model`	`Any \\| None`	`None`	Second model tried when the loop budget is exhausted
`max_iterations`	`int`	`12`	Max LLM invocations per model attempt
`max_validation_retries`	`int`	`3`	Max final answers that may fail validation
`retry_config`	`RetryConfig \\| None`	`DEFAULT_RETRY_CONFIG`	Network retry settings
`grounding_source`	`str \\| Mapping[str, str] \\| None`	`None`	What `Grounded()` fields of the final answer are checked against — see Source Grounding

Returns: tuple[ModelT | None, ExtractorRunStats]

`run_extractor_agent` — full message control¶

extract_data_with_tools builds [SystemMessage, HumanMessage] for you. When you need precise control over the conversation — multi-turn history, few-shot examples, previously gathered context — use run_extractor_agent with the same parameters but a ready-built message list:

from langchain_core.messages import HumanMessage, SystemMessage
from saidex import run_extractor_agent

messages = [
    SystemMessage(content="You are the filing assistant for project X."),
    HumanMessage(content="Earlier decision: contracts go to folder LEGAL-7."),
    HumanMessage(content="File this document: 'Service agreement with ACME...'"),
]

decision, stats = await run_extractor_agent(
    llm, FilingDecision, messages, tools=[create_folder, list_folders]
)

`final_answer_mode` — with or without tool calling¶

Value	Behaviour
`ExtractionMode.TOOL_CALLING` (default)	The schema is bound as an additional "final answer" tool; the loop ends when the LLM calls it. Most reliable.
`ExtractionMode.JSON`	Only helper tools are bound. When the LLM stops calling tools and replies with plain text, the content is parsed as JSON (tolerant of code fences, `<think>` blocks, and minor JSON damage).

Note: the helper tools always require a tool-calling-capable model — only the final answer can be collected as raw JSON. For fully tool-free extraction see Extraction Modes.

Fallback model¶

Like extract_data, the loop accepts a fallback_llm_model. When the primary model exhausts max_iterations (or max_validation_retries) without a valid final answer, the fallback model starts over with a fresh conversation built from the original messages plus a brief hint that a previous attempt failed. The returned ExtractorRunStats are the sum of both attempts with fallback_used=True.

resolution, stats = await extract_data_with_tools(
    cheap_llm,
    TicketResolution,
    task_text,
    tools=[order_status],
    fallback_llm_model=strong_llm,
)

The loop also accepts reasoning_escalation and fallback_reasoning_escalation. Reasoning levels advance with final-answer attempts only; helper-tool calls keep the current level. See Reasoning-effort escalation.

`ExtractorRunStats`¶

Field	Type	Meaning
`iterations`	`int`	Total LLM invocations in the loop
`tool_calls`	`int`	Helper-tool executions (final-answer calls not counted)
`validation_retries`	`int`	Final answers that failed schema validation
`fallback_used`	`bool`	Whether the fallback model was invoked
`salvaged`	`bool`	`True` when an `on_exhausted` hook produced the result

last_candidate / last_rejection work as in extract_data — see External Validators.

ExtractorRunStats instances support + to merge (used internally to combine primary and fallback attempts).

failure_reason uses the same FailureReason values as extract_data, plus no_tool_call, which only the agent loop produces. In the agent loop validation_exhausted is also reported when max_iterations ran out before a valid final answer; then raise max_iterations.

Synchronous usage¶

Both functions are async, but each has a synchronous wrapper — extract_data_with_tools_sync and run_extractor_agent_sync — for callers outside an event loop:

from saidex import extract_data_with_tools_sync

result, stats = extract_data_with_tools_sync(llm, MySchema, text, tools=[my_tool])

The wrappers take the same arguments and return the same (model, stats) tuple as the async versions, delegating via asyncio.run. They must not be called from within a running event loop — if one is detected they raise a clear RuntimeError. See Observability → Async and sync usage.

Observability¶

Tracing tool calls¶

When you pass callbacks=[...], the agent loop is wrapped in a single saidex.agent_loop span and every helper-tool execution is traced as its own on_tool_start / on_tool_end span (or on_tool_error if the handler raises), nested under the loop alongside the LLM generations. See Langfuse Tracing for a full walkthrough.

Context-aware validators¶

validator=with_context(fn) works in the agent loop as well. There, ValidatorContext.max_attempts is max_validation_retries + 1, and the attempt counter moves only with final-answer attempts (including malformed ones), not with helper-tool calls. If the model sends several final answers in one response, each of them may be checked with is_final=True.

def fn(inv: Invoice, ctx: ValidatorContext) -> ValidatorResult:
    if inv.total == sum(line.amount for line in inv.lines):
        return None
    return ValidationIssue(
        message="total does not match the line items",
        field_path="total",
        on_mismatch="flag" if ctx.is_final else "retry",
    )


result, stats = await extract_data_with_tools(
    llm, Invoice, text, tools=[lookup_tool], validator=with_context(fn)
)

See External Validators → Attempt context.

Agent Loop — let the LLM call your tools¶

Quick example¶

How the loop works¶

The Tool dataclass¶

extract_data_with_tools parameters¶

run_extractor_agent — full message control¶

final_answer_mode — with or without tool calling¶