跳到正文
博客园· wang_yb·· 3 小时前AI 评分62

用 Pydantic AI 从零搭建一个能干活的 AI Agent

用 Pydantic AI 从零搭建一个能干活的 AI Agent - wang_yb

AI 导读

作者用 Pydantic AI 从零搭建个人记账分析助手，通过 output_type 定义 SpendingInsight 结构化输出、用 @agent.tool 注册查询工具、以 deps_type 注入 LedgerRepo 数据类，让模型只负责理解意图，取数、聚合、环比等计算全部交给 Python。

正文

很多人第一次写 AI Agent，写完都有点失望。

代码确实能跑，模型也确实在说话，但往下走两步就露馅了：

返回的是一段字符串，你得自己 json.loads()，还得写一堆兜底防止它今天返回 JSON、明天返回 Markdown；

想让它查你自己的数据，就得手写一个 prompt 解释"我有个函数叫 xxx，参数是 yyy，你可以这样调用"，prompt 越写越长；

等你想给它写个单元测试，发现代码里到处是全局变量和一个写死的路径，根本没法测。

问题不在模型不够聪明，而在中间缺一层工程约束。

而Pydantic AI 可以干这层活：它把你熟悉的 Python 写法，类型注解、Pydantic 模型、装饰器、数据类——直接接到大模型上，让"输出必须是这个结构"，"这个函数可以被调用"，"这个数据源运行时再给"等等这些事在代码层面被约束住，而不是靠 prompt 里的恳求。

本文尝试从零搭一个个人记账分析助手：你问一句"上个月我花了多少，钱花在哪了"，它自己去读账单、算环比、拆分类、挑出异常消费，最后交回一个结构完整的结论对象。

1. 先理清概念

LLM 不等于 Agent。 调一次接口问一句答一句，那叫"用大模型"。

Agent 多出来的是它可以自己决定接下来干什么：调用你写的函数，拿到结果继续想下一步，最后交出校验过的对象。少一样，它就是个会说话的字符串生成器。

**Pydantic AI **：它是 Pydantic 团队做的 Agent 框架。

思路就一句话：用类型系统把 Agent 管起来。

你定义输出模型，它替你生成 Schema、替你校验、不过自动让模型重答；你注册一个函数，它替你翻译成模型看得懂的调用说明；你声明依赖类型，它替你在运行时把数据源送进工具。

要戒掉的坏习惯。 是那种"我先解析一下试试"的临时代码。不确定性能在框架层消掉，就别再手写 try/except 去兜。

最终要搭的东西，画出来是这样：

整条链路里真正属于"AI"的只有中间那个方块。左边是你的问题，右边是 Excel 和推送，中间读账单、算分类、找异常全是实打实的 Python。

2. 场景：记账分析助手

账单的好处是数据你自己就有。

支付宝和微信支付都能导出交易明细，整理成一张带日期、类别、金额、支付方式、备注的表（date / category / amount / method / note）就能用。

要回答的问题也就这几个：花了多少、比上个月多还是少、钱主要在哪几类、有没有不该花的。 注意这套流程是固定的——取数、聚合、算环比、拆分类、找异常；变化的是"用户想问哪段时间、关心哪个类别"。

固定的写进 Python，理解意图的交给模型，这是全文最重要的一条。

3. 先跑通：五行代码的第一个 Agent

别急着搭，先确认"模型能说话"。

python -m venv venv
source venv/bin/activate          # Windows 用 venv\Scripts\activate
pip install "pydantic-ai" pandas numpy
export OPENAI_API_KEY="sk-xxxx"   # Windows 用 setx OPENAI_API_KEY "sk-xxxx"

第一个 Agent 有两个必填项：

from pydantic_ai import Agent

from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

MODEL = OpenAIChatModel(
    "deepseek-ai/DeepSeek-V3.2",
    provider=OpenAIProvider(
        base_url="https://api.siliconflow.cn/v1",
        api_key=os.getenv("OPENAI_API_KEY"),
    ),
)

ledger_agent = Agent(
    MODEL,
    instructions=("你是一名个人财务分析助手。"),
)

result = agent.run_sync("什么是大语言模型？")
print(result.output)

我使用的是硅基流动中的 DeepSeek 模型，instructions** 是系统级人设**，对每一次运行都生效。

run_sync 是同步版，塞进 FastAPI 时换成 await agent.run(...)，API 面完全一样。

到这里 result.output 还是个字符串，能用，但没法直接进你的表格。

4. 让它交出一个对象，而不是一段话

这是收益最大的一步：**给 Agent 指定 **output_type。你把想要的结构用 Pydantic 模型描述出来，剩下的框架全包——生成 JSON Schema、和 prompt 一起发出去、拿回结果做校验、不过就带着报错让模型重答。

本文这个助手最终要交出的东西：

class MetricChange(BaseModel):
    name: str = Field(description="指标名，如 支出总额 / 笔数 / 单笔均值")
    current: float
    previous: float
    change_pct: float = Field(description="环比变化百分比，保留一位小数，单位 %")

class SpendingInsight(BaseModel):
    period: str = Field(description="分析区间，形如 2026-09-01 ~ 2026-09-30")
    headline: str = Field(description="一句话结论：花超了还是省下来了，把数字带上")
    top_categories: list[str] = Field(description="花钱最多的前三个类别，按金额降序")
    anomalies: list[str] = Field(default_factory=list, description="异常消费，没有就返回空列表",)
    suggestions: list[str] = Field(description="下个月能做的具体动作，不要写套话")
    confidence: float = Field(description="结论可信度，0 到 1 之间", ge=0.0, le=1.0)

每个字段都挂了 Field(description=...)，这不是注释，它会出现在发给模型的 Schema 里，是在明确告诉模型这个字段到底要什么，写得清楚能直接减少重试次数。

这个模型其实就是一份接口契约：定了之后，写 Excel 也好、推送到群里也好，拿到的永远是同一套字段。

5. 给它一双手：让它可以真的去查账单

光有结构化输出，模型还是不知道你上个月到底花了多少。工具就是把你的 Python 函数借给它用。

先看清一次调用的完整循环：

两个关键点：第③步和工具之间可以来回好几轮，模型会一步步补信息；而每一次工具调用都是一次模型请求，工具做得越碎，钱和时间花得越快。所以工具的设计原则是一次给足信息。

一个普通函数是怎么被模型看懂的：

装饰器的区别只有一处：@agent.tool_plain 用于不需要运行时上下文的函数；@agent.tool 第一个参数是 RunContext[...]，用来拿注入进来的东西。

@ledger_agent.tool
def get_metrics(ctx: RunContext[LedgerRepo], start: str, end: str,
                categories: list[str] | None = None) -> str:
    """查询一段日期内的支出总额、笔数、单笔均值，
       并自动与上一个等长周期做环比。

    Args:
        start: 开始日期，YYYY-MM-DD。
        categories: 消费类别列表，不传表示全部。
    """
    return json.dumps(ctx.deps.query_metrics(start, end, categories),
                      ensure_ascii=False)

那段 docstring 不是注释，是给模型看的调用说明书，写得含糊它就会在错误的时候乱调。

还有一条实操经验：所有计算都放进工具里。环比、占比、均值必须是 Python 算好再返回，模型只负责"读懂数字、说出人话"。

6. 把账单抽出去：依赖注入不是学院派

到这里代码能跑，但账单数据源是写死的。每次换文件、每次写测试都要改工具代码，改完还得祈祷没改错。依赖注入解决的就是这件事：Agent 声明自己需要什么类型，具体给哪个实例，运行时再说。

把所有取数收进一个数据类，它就是你和数据源之间的接口：

@dataclass
class LedgerRepo:
    frame: pd.DataFrame                       # 换成数据库、换成 API 都一样

    def available_dimensions(self) -> dict: ...
    def query_metrics(self, start, end, categories=None) -> dict: ...
    def breakdown_by_category(self, start, end) -> dict: ...
    def top_expenses(self, start, end, limit=5) -> dict: ...

Agent 上声明 deps_type=LedgerRepo，运行时把实例塞进去就行：

repo = LedgerRepo(frame=pd.read_csv("bill.csv", parse_dates=["date"]))
result = ledger_agent.run_sync("上个月我花超了吗？", deps=repo)
insight = result.output          # 一个 SpendingInsight 实例
print(insight.headline, insight.top_categories)

回报在写测试时立刻兑现——不用改 Agent 的任何定义，直接把依赖换成 mock：

with ledger_agent.override(deps=build_mock_repo()):
    result = ledger_agent.run_sync("看一下最近花了多少")

想连模型调用都省掉，官方还有 TestModel / FunctionModel，同样 override 掉 model 就能离线跑。

7. 两个开关：深度思考和联网

框架把可复用行为做成了可组合的 capabilities，挂上去就行，不挑供应商。

from pydantic_ai.capabilities import Thinking, WebSearch

agent = Agent(
    "deepseek-ai/DeepSeek-V3.2",
    instructions="你是一名个人财务助手，结论必须有数字支撑。",
    capabilities=[Thinking(effort="high"), WebSearch()],
)

Thinking 让它先一步步推再作答，适合"消费为什么涨"这种归因；WebSearch 给它外部视角，比如某月多花是不是赶上了节假日。

要不要开看场景：算数已经交给 Python 了，日常关掉更省钱，也免得它被网上不相干的内容带偏。

8. 自己动手试一试

完整代码：ledger_agent_demo.py: https://url11.ctfile.com/f/45455611-17569903791796-c3c2a0?p=6872 (访问密码: 6872)

有兴趣的话可以用这个代码尝试一下：

  1. 首先导入自己的 API-KEY：export OPENAI_API_KEY="sk-xxxx"
  2. 代码中的模型部分请替换成自己的模型（代码中我测试用的是硅基流动的模型）：
MODEL = OpenAIChatModel(
    "deepseek-ai/DeepSeek-V3.2",
    provider=OpenAIProvider(
        base_url="https://api.siliconflow.cn/v1",
        api_key=os.getenv("OPENAI_API_KEY"),
    ),
)
  1. 代码默认使用模拟数据，如果有自己的数据，请在环境变量中设置你的数据（CSV 格式）路径：
    csv_path = os.getenv("LEDGER_CSV")

执行结果（默认方式 python ledger_agent_demo.py）：

$ python ledger_agent_demo.py
                 pydantic-ai v2.54.0 • Python 3.12.12

      / \        agent: ledger_agent • model: openai:deepseek-ai/DeepSeek-V3.2 • output: SpendingInsight • tools: 4
     /   \         capabilities: 0
   /___.___\
  /    |    \    observability: off — see every model and tool call live, with cost
/      |      \    set it up free with Logfire and a GitHub login: https://pydantic.dev/ai-setup.md
`---.._|_..---'    or use any OpenTelemetry backend: https://pydantic.dev/docs/ai/logfire/#otel

                 goes away once observability is on — or PYDANTIC_AI_NO_BANNER=1

== 上个月花超了，总支出 16,864.79 元，比上个月多花 67.1%
-- 区间： 2026-09-01 ~ 2026-09-30
   支出总额: 16864.79（环比 67.1%）
   笔数: 243.0（环比 37.3%）
   单笔均值: 69.4（环比 21.7%）
-- 花钱最多的类别： 购物、餐饮、娱乐
-- 异常消费：
   - 2026-09-01 居住类支出 2,439.86 元
-- 原因分析：
   - 购物支出 4,825.27 元占总支出 28.61%，比上个月明显增加
   - 9月初有大额居住支出 2,439.86 元一次性拉高整体消费
   - 餐饮消费 3,870.85 元笔数达 90 笔，单日消费频次较高
-- 下个月怎么做：
   - 下个月重点关注购物类消费，把单笔购物控制在 200 元以内
   - 餐饮支出过高，尝试工作日午餐自带或降低外卖频次
   - 大额居住消费已发生，下个月房租支出会恢复正常
-- 可信度： 0.95

如果使用自己的数据，请先导出自己的交易明细，整理成带 date / category / amount / method / note 五列的 CSV（比如文件名 bill.csv）；

然后 LEDGER_CSV=./bill.csv python ledger_agent_demo.py。

来源：博客园 · cnblogs.com

Agent 智能体部署工程MCP 与工具调用教程实践#教程/实践#Agent#MCP/工具调用#部署/工程