The Agent Framework Landscape

# script_id: day_057_agent_frameworks/pydantic_minimal
# Install: uv add "pydantic-ai-slim[openai]==2.66.0"  ·  Env: none (TestModel)  ·  Services: none
import os

os.environ.setdefault("PYDANTIC_AI_NO_BANNER", "1")  # keep the startup banner out of the output
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")

class IssueFinding(BaseModel):
    summary: str
    doc_ids: list[str]
    duplicate_of: int | None = None

# defer_model_check=True: build the agent without OPENAI_API_KEY; the model is resolved at run time
agent = Agent(f"openai:{MODEL}", output_type=IssueFinding,
              instructions="Diagnose the Orbit issue.", defer_model_check=True)
result = agent.run_sync("Issue #240: nightly job ran twice on DST fall back", model=TestModel())
print(repr(result.output))
print(result.usage.requests, "request(s)")

IssueFinding(summary='a', doc_ids=['a'], duplicate_of=None)
1 request(s)

# script_id: day_057_agent_frameworks/silent_defaults
# Install: uv add "pydantic-ai-slim[openai]==2.66.0"  ·  Env: none  ·  Services: none
import os
import warnings

os.environ["PYDANTIC_AI_NO_BANNER"] = "1"
os.environ.pop("OPENAI_API_KEY", None)  # simulate a CI box or a fresh laptop
from pydantic import BaseModel
from pydantic_ai import Agent, UserError
from pydantic_ai.models.test import TestModel

class IssueFinding(BaseModel):
    summary: str

try:  # 1. building an OpenAI agent without a key fails at construction, not at the first call
    Agent("openai:gpt-6-luna", output_type=IssueFinding)
except UserError as e:
    print("1.", type(e).__name__, str(e)[:70])

try:  # 2. tutorials written for the 0.x API use result_type=
    Agent(TestModel(), result_type=IssueFinding)
except TypeError as e:
    print("2.", e)

result = Agent(TestModel(), output_type=IssueFinding).run_sync("hi")
try:  # 3. ...and result.data
    result.data
except AttributeError as e:
    print("3.", e)

with warnings.catch_warnings(record=True) as caught:  # 4. usage is a property now
    warnings.simplefilter("always")
    result.usage()
    print("4.", caught[0].message)

1. UserError Set the `OPENAI_API_KEY` environment variable or pass it via `OpenAIPr
2. Agent.__init__() got an unexpected keyword argument 'result_type'
3. 'AgentRunResult' object has no attribute 'data'
4. `usage` is a property, not a method: use `.usage` instead of `.usage()`.

# script_id: day_057_agent_frameworks/model_id_check
# fragment: reference shapes; llama-index and crewai are not installed in the course venv
MODEL = "gpt-6-luna"
from langchain_openai import ChatOpenAI
chat = ChatOpenAI(model=MODEL, use_responses_api=True)    # never ChatOpenAI(): default is gpt-3.5-turbo

from crewai import LLM
crew_llm = LLM(model=f"openai/{MODEL}", api="responses")  # default api is "completions"

from llama_index.llms.openai import OpenAI as LlamaOpenAI
llm = LlamaOpenAI(model="gpt-6.1-terra")   # the constructor succeeds...
llm.metadata                               # ...first use raises ValueError: Unknown model ...

from openai import OpenAI                  # framework-free startup check: is the ID live?
live = {m.id: m.shutdown_date for m in OpenAI().models.list()}
assert MODEL in live, f"{MODEL} is not available to this key"

# script_id: day_057_agent_frameworks/pydantic_agent
# atlas/pydantic_agent.py: typed Orbit issue agent (PydanticAI)
# Install: uv add "pydantic-ai-slim[openai]==2.66.0"
# Env: OPENAI_API_KEY for live runs only; ATLAS_MODEL, ORBIT_DATA optional  ·  Services: none
from __future__ import annotations

import os
import re
import sqlite3
from contextlib import closing
from dataclasses import dataclass, field
from pathlib import Path

os.environ.setdefault("PYDANTIC_AI_NO_BANNER", "1")
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext, UsageLimitExceeded, UsageLimits

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))

@dataclass
class MockTracker:  # built on Day 37: side effects are recorded, never sent
    calls: list[dict] = field(default_factory=list)
    def record(self, action: str, **args: object) -> None:
        self.calls.append({"action": action, **args})

@dataclass
class Deps:
    db_path: Path
    tracker: MockTracker
    docs_dir: Path = DATA_DIR / "docs"
    current_issue: int | None = None  # tools only see issues filed before this one

class IssueFinding(BaseModel):
    summary: str = Field(description="One or two sentences: likely cause and next step")
    doc_ids: list[str] = Field(description="Doc ids (file names without .md) that support it")
    duplicate_of: int | None = Field(default=None, description="Earlier issue id with the same cause")

def connect_ro(db_path: Path) -> sqlite3.Connection:  # built on Day 33: read-only by URI
    return sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)

# script_id: day_057_agent_frameworks/pydantic_agent
def search_issues(ctx: RunContext[Deps], query: str, limit: int = 5) -> list[dict]:
    """Search earlier Orbit issues by keyword in the title or body.

    Args:
        query: a keyword or short phrase, e.g. "ran twice" or "ORB-1402".
        limit: maximum number of issues to return (at most 10).
    """
    before = ctx.deps.current_issue or 10**9
    with closing(connect_ro(ctx.deps.db_path)) as con:
        rows = con.execute(
            "SELECT id, title, state, duplicate_of FROM issues "
            "WHERE (title LIKE ?1 OR body LIKE ?1) AND id < ?2 ORDER BY id LIMIT ?3",
            (f"%{query}%", before, min(limit, 10)),
        ).fetchall()
    return [dict(zip(("id", "title", "state", "duplicate_of"), r)) for r in rows]

def search_docs(ctx: RunContext[Deps], query: str, limit: int = 3) -> list[dict]:
    """Find Orbit doc pages that mention every word of the query. Returns doc ids and snippets."""
    words = [w.lower() for w in re.findall(r"[\w-]+", query)]
    hits = []
    for path in sorted(ctx.deps.docs_dir.glob("*.md")):
        text = path.read_text(encoding="utf-8")
        if all(w in text.lower() for w in words):
            hits.append({"doc_id": path.stem, "snippet": text[:300]})
    return hits[: min(limit, 5)]

# script_id: day_057_agent_frameworks/pydantic_agent
INSTRUCTIONS = (
    "You triage new issues for Orbit, a Python background-jobs library. Search earlier issues "
    "and the docs before answering. Cite only doc ids returned by search_docs. Set duplicate_of "
    "only when an earlier issue has the same root cause."
)
PRICE_PER_M = {"gpt-6-luna": (0.10, 0.50)}  # USD in/out per 1M tokens, as of 2026-11; verify

def build_pydantic_agent(model: str | None = None) -> Agent[Deps, IssueFinding]:
    return Agent(model or f"openai:{MODEL}", deps_type=Deps, output_type=IssueFinding,
                 instructions=INSTRUCTIONS, tools=[search_issues, search_docs],
                 retries=2, defer_model_check=True)

@dataclass
class AgentResult:  # built on Day 48 (compact: steps is a count here)
    text: str
    stop_reason: str
    steps: int
    usage: dict
    cost_usd: float
    partial: bool = False

def run_pydantic_agent(task: str, deps: Deps, *, agent: Agent | None = None,
                       model: object = None, max_steps: int = 8) -> AgentResult:
    agent = agent or build_pydantic_agent()
    try:
        r = agent.run_sync(task, deps=deps, model=model,
                           usage_limits=UsageLimits(request_limit=max_steps))
    except UsageLimitExceeded as e:  # Day 48's max_steps stop, surfaced as a partial result
        return AgentResult(str(e), "max_steps", max_steps, {}, 0.0, partial=True)
    u = r.usage
    p_in, p_out = PRICE_PER_M.get(MODEL, (0.0, 0.0))
    return AgentResult(text=r.output.model_dump_json(), stop_reason="completed", steps=u.requests,
                       usage={"input_tokens": u.input_tokens, "output_tokens": u.output_tokens,
                              "tool_calls": u.tool_calls},
                       cost_usd=(u.input_tokens * p_in + u.output_tokens * p_out) / 1_000_000)

# script_id: day_057_agent_frameworks/pydantic_agent
if __name__ == "__main__":  # keyless smoke run; pass model=None (and set OPENAI_API_KEY) to go live
    from pydantic_ai.models.test import TestModel

    deps = Deps(db_path=DATA_DIR / "issues" / "orbit.db", tracker=MockTracker(), current_issue=240)
    print(run_pydantic_agent("Issue #240: nightly job ran twice on DST", deps, model=TestModel()))

AgentResult(text='{"summary":"a","doc_ids":["a"],"duplicate_of":null}', stop_reason='completed', steps=2, usage={'input_tokens': 345, 'output_tokens': 28, 'tool_calls': 2}, cost_usd=4.85e-05, partial=False)

# script_id: day_057_agent_frameworks/test_pydantic_agent
# tests/test_pydantic_agent.py · Install: uv add pytest "pydantic-ai-slim[openai]==2.66.0"
# Env: none · Needs pydantic_agent.py (this lesson) next to it; in Atlas: from atlas.pydantic_agent import ...
import pytest
import pydantic_ai.models
from pydantic_ai.messages import ModelMessage, ModelResponse, ToolCallPart, ToolReturnPart
from pydantic_ai.models.function import AgentInfo, FunctionModel
from pydantic_ai.models.test import TestModel
from pydantic_agent import DATA_DIR, Deps, IssueFinding, MockTracker, build_pydantic_agent

pydantic_ai.models.ALLOW_MODEL_REQUESTS = False  # a forgotten override fails instead of billing

@pytest.fixture
def deps() -> Deps:
    return Deps(db_path=DATA_DIR / "issues" / "orbit.db", tracker=MockTracker(), current_issue=240)

def tool_calls(result) -> list[str]:
    return [p.tool_name for m in result.all_messages() for p in m.parts if isinstance(p, ToolCallPart)]

def test_wiring_with_testmodel(deps: Deps) -> None:
    agent = build_pydantic_agent()
    with agent.override(model=TestModel()):
        result = agent.run_sync("Issue #240: job ran twice", deps=deps)
    assert {"search_issues", "search_docs"} <= set(tool_calls(result))
    assert isinstance(result.output, IssueFinding)
    assert deps.tracker.calls == []  # a read-only agent records no side effects

# script_id: day_057_agent_frameworks/test_pydantic_agent
def search_then_answer(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    returns = [p for m in messages for p in m.parts if isinstance(p, ToolReturnPart)]
    if not returns:  # step 1: search past issues
        return ModelResponse(parts=[ToolCallPart("search_issues", {"query": "ORB-1402"})])
    first_hit = returns[0].content[0]["id"]  # step 2: answer via the output tool
    out = {"summary": "DST fall-back double run.", "doc_ids": ["scheduling-cron"],
           "duplicate_of": first_hit}
    return ModelResponse(parts=[ToolCallPart(info.output_tools[0].name, out)])

def test_scripted_two_steps(deps: Deps) -> None:
    agent = build_pydantic_agent()
    with agent.override(model=FunctionModel(search_then_answer)):
        result = agent.run_sync("Issue #240: job ran twice on DST", deps=deps)
    assert tool_calls(result)[0] == "search_issues"
    assert result.output.duplicate_of == 59  # issue 59 is the earliest ORB-1402 report
    assert result.usage.requests == 2

if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q", "-p", "no:cacheprovider"]))

..                                                                       [100%]
2 passed in 0.03s

# script_id: day_057_agent_frameworks/compare_task
# Install: uv add "pydantic-ai-slim[openai]==2.66.0"  ·  Needs pydantic_agent.py (this lesson)
# Env: OPENAI_API_KEY (optional: without it, runs on TestModel), ATLAS_MODEL, ORBIT_DATA
import json
import os
from contextlib import closing
from pathlib import Path

from pydantic_ai.models.test import TestModel
from pydantic_agent import DATA_DIR, MODEL, Deps, MockTracker, connect_ro, run_pydantic_agent

ISSUES = [22, 41, 102, 149, 240]  # the five-issue scoreboard set (three have a gold duplicate)
LIVE = bool(os.environ.get("OPENAI_API_KEY"))
DB = DATA_DIR / "issues" / "orbit.db"

def load_issue(issue_id: int) -> tuple[str, int | None]:
    with closing(connect_ro(DB)) as con:
        title, body, dup = con.execute(
            "SELECT title, body, duplicate_of FROM issues WHERE id = ?", (issue_id,)).fetchone()
    return f"Issue #{issue_id}: {title}\n\n{body[:1500]}", dup

src = Path(__file__).with_name("pydantic_agent.py").read_text(encoding="utf-8").splitlines()
loc = sum(1 for ln in src if ln.strip() and not ln.lstrip().startswith("#"))
print(f"mode: {'live ' + MODEL if LIVE else 'offline TestModel'} · agent LoC: {loc}")
print(f"{'issue':>5} {'steps':>5} {'tools':>5} {'tokens':>7} {'cost $':>9}  dup pred/gold")
total = 0.0
for issue_id in ISSUES:
    task, gold = load_issue(issue_id)
    deps = Deps(db_path=DB, tracker=MockTracker(), current_issue=issue_id)
    res = run_pydantic_agent(task, deps, model=None if LIVE else TestModel())
    pred = json.loads(res.text).get("duplicate_of") if not res.partial else "-"
    tokens = res.usage.get("input_tokens", 0) + res.usage.get("output_tokens", 0)
    total += res.cost_usd
    print(f"{issue_id:>5} {res.steps:>5} {res.usage.get('tool_calls', 0):>5} {tokens:>7} "
          f"{res.cost_usd:>9.5f}  {pred}/{gold}")
print(f"total ≈ ${total:.4f} (prices as of 2026-11; verify)")

mode: offline TestModel · agent LoC: 88
issue steps tools  tokens    cost $  dup pred/gold
   22     2     2     539   0.00007  None/4
   41     2     2     593   0.00007  None/4
  102     2     2     851   0.00010  None/None
  149     2     2     577   0.00007  None/None
  240     2     2     667   0.00008  None/59
total ≈ $0.0004 (prices as of 2026-11; verify)