Lesson 14 — bounded agent behaviour

What you will build

The thing this path has been walking towards, stated plainly: a loop that calls a model, lets it ask for tools, and stops. There is no agent loop in the runtime. What exists is chat_with_tools — a bounded tool loop with a round limit — and the workflow engine, which supplies the deadline and the escalation path. An agent here is those two things composed, plus a budget someone chose.

What you need

Nothing new.

examples/learn_ai/ai14_agent.py

"""Lesson 14 — bounded agent behaviour: a loop, a budget, and an escalation.

Run it with:

    python -m learn_ai.ai14_agent
"""

from __future__ import annotations

import asyncio
from collections.abc import Mapping
from dataclasses import dataclass

from xyberos import StepContext, Workflow, seconds
from xyberos.ai import (
    AIFacade,
    ChatRequest,
    FakeModelProvider,
    Message,
    ModelReply,
    Tool,
    ToolCall,
    ToolSet,
    ToolSpec,
)
from xyberos.errors import AIError

TICKETS = {"T-100": "printer is on fire"}

LOOKUP_SCHEMA = {
    "type": "object",
    "properties": {"ticket": {"type": "string"}},
    "required": ["ticket"],
    "additionalProperties": False,
}


@dataclass(frozen=True)
class Budget:
    """What one run of the agent may spend. Every number here is a bound, not a target."""

    rounds: int = 2
    deadline: float = 5.0


def lookup_ticket() -> Tool:
    """The only capability the agent has: reading one ticket."""

    async def handler(arguments: Mapping[str, object], context: object) -> dict[str, object]:
        return {"title": TICKETS.get(str(arguments["ticket"]), "no such ticket")}

    return Tool(
        spec=ToolSpec(
            name="lookup_ticket",
            description="Look up one ticket's title",
            parameters=LOOKUP_SCHEMA,
        ),
        handler=handler,
    )


def agent_step(assistant: AIFacade, tools: ToolSet, budget: Budget) -> Workflow:
    """One step that runs a bounded tool loop, and escalates rather than guessing."""

    async def run(value: str, context: StepContext) -> str:
        try:
            response = await assistant.chat_with_tools(
                ChatRequest(
                    messages=(
                        Message.system("Answer using the tools. If you cannot, say you cannot."),
                        Message.user(value),
                    )
                ),
                tools=tools,
                max_rounds=budget.rounds,
            )
        except AIError as error:
            return f"escalated: {type(error).__name__}"
        return response.message.content

    return Workflow("agent", version="1", input_type=str).step(
        "answer", run, timeout=seconds(budget.deadline)
    )


async def main() -> None:
    """Two runs: one the agent answers with a tool, one it escalates."""
    tools = ToolSet.of(lookup_ticket())

    answered = FakeModelProvider(
        [
            ModelReply.tools(ToolCall(name="lookup_ticket", arguments={"ticket": "T-100"})),
            ModelReply.ok("T-100 is 'printer is on fire'."),
        ]
    )
    first = await agent_step(AIFacade(answered), tools, Budget()).run(
        input="What is ticket T-100?", timeout=seconds(10)
    )
    print("asked once:   ", first.value)

    looping = FakeModelProvider(
        [ModelReply.tools(ToolCall(name="lookup_ticket", arguments={"ticket": "T-100"}))] * 4
    )
    second = await agent_step(AIFacade(looping), tools, Budget(rounds=1)).run(
        input="What is ticket T-100?", timeout=seconds(10)
    )
    print("asked forever:", second.value)

    refusing = FakeModelProvider([ModelReply.refusal(content="I cannot help with that.")])
    third = await agent_step(AIFacade(refusing), tools, Budget()).run(
        input="What is ticket T-100?", timeout=seconds(10)
    )
    print("refused:      ", third.value)
    print("the loop is chat_with_tools, bounded by max_rounds; the runtime has no agent loop")


if __name__ == "__main__":
    asyncio.run(main())

Budget is a dataclass with two numbers, and it is the most important object on this page: a round limit and a deadline. Every agent story that goes wrong goes wrong by removing one of them.

The step's error handling is the second decision: an AIError — a round limit, a refusal, a timeout, a provider that is down — becomes an escalation, not a retry and not an invented answer.

How to run it

PowerShellPOSIX

$env:PYTHONPATH = "src;examples"
python -m learn_ai.ai14_agent

PYTHONPATH=src:examples python -m learn_ai.ai14_agent

What it prints

asked once:    T-100 is 'printer is on fire'.
asked forever: escalated: AIToolError
refused:       escalated: AIRefusalError
the loop is chat_with_tools, bounded by max_rounds; the runtime has no agent loop

Three runs, three answers, and two of them are escalations for different reasons:

Run	Outcome	Why escalating is right
answers with a tool	T-100 is 'printer is on fire'.	the loop finished inside its budget
asks forever	escalated: AIToolError	the model kept asking; the round limit is what stopped it
refuses	escalated: AIRefusalError	the model declined; retrying would cost tokens and refuse again

The last line printed is the honest description of the architecture: the loop lives in chat_with_tools, the budget lives in your code, and the sequencing lives in a workflow. Nothing in the runtime decides for you how long an agent may think.

Common mistakes

• Removing the round limit because it was inconvenient. The limit is the difference between an agent and a runaway loop that bills by the minute.
• Letting the agent decide when to escalate. Escalation is a policy: an error class, a confidence, a budget — decided by the application, and testable without a model.
• Giving the loop a write tool. With tools, the model's output becomes actions; every action needs an approval boundary, which is the hybrid path's whole subject.
• Treating the model's summary as the record. The tool events and the step outcomes are the record; the answer is what the user was told.
• An unbounded conversation as "context". A tool loop plus an unbounded window is the most expensive request in the system; bound both, and say what the bound costs.

Next

Lesson 15 turns everything so far into tests that run with no model service — which is how this path itself is verified.