《AI Agent 高级编程》2.3 与 minagent 对照：代码量与可测性对比

from __future__ import annotations
import json
from typing import Any
from minagent.types import Message, ModelResponse, ToolCall
from minagent.tools import Tool, ToolRegistry
from minagent.fake import FakeModel
from minagent.agent import Agent

CORPUS = {"pg": "ResearchCopilot 使用 PostgreSQL 17.11。",
          "cache": "语义缓存用 Redis 7.4。"}


def search(query: str, k: int = 2) -> list[str]:
    """按关键词检索本地语料。"""
    q = query.lower()
    hits = [f"[{key}] {text}" for key, text in CORPUS.items() if q in key or q in text.lower()]
    return hits[:k]


SEARCH_SCHEMA: dict[str, Any] = {
    "type": "object",
    "properties": {"query": {"type": "string"}, "k": {"type": "integer"}},
    "required": ["query"],
}


def build() -> tuple[Agent, FakeModel]:
    reg = ToolRegistry()
    reg.add(Tool(name="search", description="按关键词检索语料",
                 parameters=SEARCH_SCHEMA, func=search))
    script = [
        [ToolCall(id="c1", name="search", arguments={"query": "pg"})],
        "依据 [pg]：ResearchCopilot 使用 PostgreSQL 17.11。",
    ]
    model = FakeModel(script)
    return Agent(model, reg, system="先检索再回答，保留引用。"), model


agent, model = build()
result = agent.run("用的哪个 Postgres？")
print("final:", result.final)
print("steps:", [(s.kind, s.detail) for s in result.steps])
print("模型第 1 次看到的工具:", [t["name"] for t in model.calls[0][1]])

final: 依据 [pg]：ResearchCopilot 使用 PostgreSQL 17.11。
steps: [('tool_call', 'search -> ["[pg] ResearchCopilot 使用 PostgreSQL 17.11。"]'), ('final', '依据 [pg]：ResearchCopilot 使用 PostgreSQL 17.11。')]
模型第 1 次看到的工具: ['search']

from __future__ import annotations
import os
os.environ["PYDANTIC_AI_NO_BANNER"] = "1"
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.function import FunctionModel, AgentInfo
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart, ToolCallPart

CORPUS = {"pg": "ResearchCopilot 使用 PostgreSQL 17.11。",
          "cache": "语义缓存用 Redis 7.4。"}


class FakeRetriever:
    def __init__(self, corpus: dict[str, str]) -> None:
        self.corpus = corpus
        self.calls: list[str] = []

    def search(self, query: str, k: int = 2) -> list[str]:
        self.calls.append(query)
        q = query.lower()
        hits = [f"[{key}] {text}" for key, text in self.corpus.items() if q in key or q in text.lower()]
        return hits[:k]


@dataclass
class Deps:
    retriever: FakeRetriever


agent = Agent("test", deps_type=Deps, system_prompt="先检索再回答，保留引用。")


@agent.tool
def search(ctx: RunContext[Deps], query: str) -> list[str]:
    """按关键词检索本地语料。"""
    return ctx.deps.retriever.search(query)


def scripted(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    returns = [p for p in messages[-1].parts if p.part_kind == "tool-return"]
    if not returns:
        return ModelResponse(parts=[ToolCallPart(tool_name="search", args={"query": "pg"})])
    return ModelResponse(parts=[TextPart(content=f"依据检索：{returns[0].content}")])


retriever = FakeRetriever(CORPUS)
res = agent.run_sync("用的哪个 Postgres？", deps=Deps(retriever), model=FunctionModel(scripted))
print("output:", res.output)
print("检索器调用:", retriever.calls)
print("工具调用记录:", [p.tool_name for m in res.all_messages()
                    for p in m.parts if p.part_kind == "tool-call"])

output: 依据检索：['[pg] ResearchCopilot 使用 PostgreSQL 17.11。']
检索器调用: ['pg']
工具调用记录: ['search']

assert [t["name"] for t in model.calls[0][1]] == ["search"]     # 模型看到的工具
assert result.steps[-1].kind == "final"                          # 收敛，不是撞 max_steps
assert result.final == "依据 [pg]：ResearchCopilot 使用 PostgreSQL 17.11。"

calls = [p.tool_name for m in res.all_messages()
         for p in m.parts if p.part_kind == "tool-call"]
assert calls == ["search"]                                       # 工具调用记录
assert retriever.calls == ["pg"]                                 # 业务侧探针
assert "PostgreSQL 17.11" in res.output                          # 终值