《AI Agent 高级编程》2.1 核心模型：Agent/RunContext/依赖注入重写问答路径

from pydantic_ai import Agent

agent = Agent(
    "test",                                   # 模型名；本节用假模型占位
    deps_type=Deps,                           # 声明「运行时要注入的依赖」类型
    system_prompt="先检索再回答，回答里保留引用。",  # 系统提示词
)

import inspect
from pydantic_ai import Agent

print([p for p in inspect.signature(Agent.__init__).parameters][:6])
# ['self', 'model', 'output_type', 'instructions', 'system_prompt', 'deps_type']

import os
os.environ["PYDANTIC_AI_NO_BANNER"] = "1"
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel


@dataclass
class Deps:
    tenant: str
    retriever: str


agent = Agent("test", deps_type=Deps)


@agent.tool
def whoami(ctx: RunContext[Deps]) -> str:
    """返回本次 run 注入的依赖摘要。"""
    return f"tenant={ctx.deps.tenant} retriever={ctx.deps.retriever}"


print(agent.run_sync("who", deps=Deps(tenant="acme", retriever="fake-retriever"),
                     model=TestModel()).output)

{"whoami":"tenant=acme retriever=fake-retriever"}

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（容器 ab-pg，端口 15435）。",
    "cache": "语义缓存用 Redis 7.4（容器 ab-redis，端口 16380）。",
    "loop": "引擎停止条件只看 tool_calls 是否为空。",
}


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 text.lower() or any(tok in text.lower() for tok in q.split())]
        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": "postgres"})])
    return ModelResponse(parts=[TextPart(content=f"依据检索：{returns[0].content}")])


retriever = FakeRetriever(CORPUS)
result = agent.run_sync("ResearchCopilot 用的 Postgres 是哪个版本？",
                        deps=Deps(retriever), model=FunctionModel(scripted))
print("output:", result.output)
print("检索器收到:", retriever.calls)
print("消息数:", len(result.all_messages()))
print("工具调用:", [p.tool_name for m in result.all_messages()
                    for p in m.parts if p.part_kind == "tool-call"])

output: 依据检索：['[pg] ResearchCopilot 使用 PostgreSQL 17.11（容器 ab-pg，端口 15435）。']
检索器收到: ['postgres']
消息数: 4
工具调用: ['search']

from pydantic_ai.models.test import TestModel

smoke = agent.run_sync("用的哪个 Postgres？", deps=Deps(FakeRetriever(CORPUS)),
                       model=TestModel())
print("smoke output:", smoke.output[:60])

smoke output: {"search":["[pg] ResearchCopilot 使用 PostgreSQL 17.11（容器 ab-p