PydanticAI生产级Agent实战:从类型安全到可观测性

📅 发布时间:2026/10/21 6:55:21
PydanticAI生产级Agent实战:从类型安全到可观测性
做LLM应用做到后期我最怕的不是模型答错而是模型答得“看起来对但没法用”。你说返回JSON它给你多出几个字段你要的是枚举值它给你一句“很高”你说这是0到1的置信度它传了个字符串。每一处偏差到生产环境都是线上事故。把大模型接进业务系统本质是在跟不确定性打交道而工程上对抗不确定性的手段从来都是类型、校验、契约和可观测性。这篇文章是系列实战项目里的第五个项目主题很明确基于PydanticAI这种类型优先的Agent框架实现一个真正能扛线上压力的生产级LLM Agent。我会把从架构思考到代码落地的全过程摊开讲适合已经跑通过模型API、想进一步把代码做成企业级工程的Python开发者。1. 为什么生产级Agent必须解决类型安全1.1 从“能跑通到能上线”之间隔着一堆隐形坑在Demo阶段模型输出的文本只要能贴出来看一眼项目就算“成了”。但到了生产环境Agent的输出会被下游系统消费它会作为工单系统的分类字段、作为数据库里的一条记录、作为另一个Agent的输入。这时候输出就不是“一句话”而是一份必须遵守契约的数据。我见过太多场景翻车。要求模型返回JSON结果字段名变成首字母大写的驼峰对不上要求category只能是“网络、账号、计费、其他”模型返回了“系统问题”这种新增值要求confidence是0到1之间的浮点数模型返回了一个百分比字符串“85%”。更隐蔽的是模型偶尔多返回一两个字段当时解析看着没事后续代码版本升级却突然崩掉。这些问题的根源在于自然语言生成天然具有随机性而业务系统天然要求确定性。prompt里写的“请严格按照格式”只是软约束它经不起边界情况和模型版本迭代的考验。生产系统需要的是硬约束在进程边界处对输入输出做一次全面校验不合法就直接拒绝、重试或降级而不是把脏数据放进行业核心流程流转。这里有一个关键认知转变LLM生成的输出本质上等同于一个外部服务返回的响应体。你调用别的HTTP接口时会用response model做反序列化和校验为什么调用大模型时能容忍它裸返回一个字符串然后到处切片处理类型安全就是把这一层校验前移、内建到Agent框架里让每一份从模型侧出来的数据都先过一遍合同再进业务代码。1.2 PydanticAI 为什么把类型安全做成地基PydanticAI最让我欣赏的一点是它没有把类型安全当作“附加功能”而是直接做成了框架的地基。Pydantic本身是Python里最流行的数据校验库之一定义模型类、自动生成JSON Schema、执行字段校验和类型转换这一套成熟能力被直接继承到了Agent框架里。使用起来感受完全不同。你定义Agent的输出类型不是写死一个巨长的prompt让它猜而是声明一个Pydantic模型类字段约束、枚举、嵌套结构一目了然。模型返回后框架会按这个模型做解析和校验校验失败时可以触发重试机制让模型重新生成。工具函数同样受到类型约束框架会从函数签名生成工具Schema模型在调用工具时参数也会经过校验。这和很多“编排优先”的框架是两条路线。编排优先的框架会提供一大堆链式组件、记忆模块、插件市场帮你把业务串起来类型校验只是其中一环甚至不存在。PydanticAI的路线更像是把一个Agent当成一个强类型函数来对待输入是结构化的用户消息和依赖输出是校验过的结构化对象。类比一下普通框架让你把大模型当“同事”你说一句它做一句结果靠人情维系PydanticAI让你把大模型当“外部服务”前后端签好接口文档任何违约都在网关层拦截。后者才是企业级项目需要的稳定关系。尤其对于从Java这类强类型语言转过来的同学你会觉得Agent开发终于有章法了而不是在写一段充满咒语的脚本。1.3 企业级工程诉求与框架能力的映射我在实际规划生产级Agent时习惯先列一份工程诉求清单再对照框架能力看哪些被天然满足哪些还需要自己补。这张表是我自己整理的常用对照供参考工程诉求说明PydanticAI的对应能力输出契约校验字段类型、枚举范围、嵌套结构不能漂移结果模型解析与校验验证失败自动重试工具调用安全模型不能把工具参数拼错类型工具函数签名自动生成Schema参数强校验可观测性每次请求的token、耗时、重试次数可查运行结果对象、事件钩子、请求上下文可测试性不依赖真实模型也能做单测可注入依赖、结构化输出可直接断言配置治理模型、提示词、密钥分开管理模型标识走配置提示词版本化靠工程约束不是说框架能解决所有问题而是说它把最难的那部分——不可预测的输出——放到了一个可校验、可失败的通道里。我在项目五里最终选它核心原因是业务系统对输入输出有强约束分类必须是预置枚举紧急度要有明确映射置信度要参与后续自动决策。这要求框架不仅能跑通对话还要能产出信得过、查得到、可回滚的数据。选型时我也对比过几个方向有的输出校验要靠自己额外封装有的工具调用链路是黑盒到了排查线上问题的时候你根本不知道模型那一步做了什么决定。PydanticAI的模型和工具链路相对透明出错时还能拿到原始消息与最终输出这对排障特别重要。2. 框架的整体架构设计与模块边界2.1 一个Agent单元的职责划分先不谈代码谈组织架构。生产级的Agent应用绝对不是把一堆Prompt拼在一起就完事。我倾向于把系统拆成若干“Agent单元”每个单元负责一个有明确边界的业务动作对外暴露结构化输入输出内部可以有自己的工具和上下文。一个标准Agent单元有四个组成部分。模型配置选用哪个模型、温度参数、最大重试次数系统提示词描述角色、任务规则、输出约束工具集允许该Agent调用的函数列表每个工具都有签名和说明结果类型当前任务必须返回的结构化数据类型。设计边界时有个原则如果某个Agent要挂超过10个工具或者系统提示词超过一屏说明职责过重要拆。举个例子我在做一个工单处理系统时不会用一个巨大的Agent同时做分类、紧急度评估、知识库检索、回复生成。我会拆成三个工单分诊Agent负责分类和紧急度判断输出结构化结果方案检索Agent基于分类结果检索知识库返回候选方案回复生成Agent把方案结合工单原文生成一段给用户的回复。每个Agent只做一个决策决策结果用类型固定的对象传递。这样做最大的好处是单个Agent出错你很容易知道错在哪一环而且可以针对不同环节用不同的模型。重要的环节用贵而准的模型简单环节用便宜快的模型成本控制也更容易做。2.2 从单Agent到多Agent的组织形态多个Agent之间怎么协作是架构里必须提前想清楚的事。一般有三种常见形态流水线模式A的输出作为B的输入像管道一样顺序往下走适合分类、抽取、整理这类任务流路由模式由路由Agent根据输入派发到不同的专业Agent适合客服机器人这种入口统一、后台能力分域的场景会合模式多个Agent并行处理子问题最后汇总Agent合并结果适合需要多角度分析的场景。我更推荐流水线和路由的组合。因为每个Agent都输出强类型对象Agent之间的数据传递天然是结构化的不会出现“上一环节生成了一段废话下一环节还得自己解析”的尴尬。多Agent之间的状态也可以统一存在结构化数据里方便追踪和审计。有一点要特别注意别让Agent之间直接传递自由文本并反复对话。这样既难以控制成本又很容易出现上下文漂移。正确做法是把上一步的结构化输出当作下一步的输入消息配合系统提示词说明“你将收到上一步的结论基于这些结论做后续处理”。一旦消息退回自由文本你前面做的类型安全设计就白费了。2.3 依赖注入与上下文处理的思路Agent要从业务系统里读取数据就得访问外部服务。生产环境里模型服务、数据库连接、内部接口调用这些都是Agent的外部依赖。如果依赖写在工具函数内部测试和替换会非常痛苦。PydanticAI的依赖注入机制解决了这个痛点。你可以定义Deps类型把连接、配置、服务客户端封装进去在运行Agent时一次性传入。工具函数通过上下文对象获取依赖而不是自己创建全局单例。这样做之后Agent的核心逻辑和外部资源解耦换数据库、换安全服务、换配置中心都不用改Agent本体。依赖的生命周期也要分清楚。应用级依赖比如配置、日志客户端、数据库连接池进程启动时创建全局共享请求级依赖比如当前用户身份、本次请求ID、会话上下文每次请求时创建避免串数据。我用一个比较土但很有效的办法检查代码是否健康看工具函数里有没有出现模块级可变全局变量。只要出现大概率就是并发安全隐患。依赖注入还有一个隐藏福利测试。单测时传入内存版依赖集成测试时传入真实依赖Agent代码本身不用改。这在后面讲测试策略时会重点展开。3. 核心代码实现与关键步骤3.1 环境准备与最小可运行示例先搭环境。我习惯用虚拟环境隔离项目依赖然后安装PydanticAI。这里以我当时使用的版本为例版本更新后API可能有微调具体以官方文档为准。pip install pydantic-aiPydanticAI会依赖Pydantic v2安装时会自动带上。接着配置模型供应商的访问信息一般通过环境变量传入。我在项目里会统一做一个配置组件从环境变量读取模型名称和密钥不会在业务代码里散落任何凭据。一个最小可运行的Agent大概长这样import asyncio from pydantic_ai import Agent agent Agent( your-model-identifier, # 替换为你接入的模型服务商约定标识 system_prompt你是一个简洁的助手直接回答问题。, ) async def main(): result await agent.run(用一句话解释什么是生产级Agent) print(result.output) if __name__ __main__: asyncio.run(main())这个示例虽然简单但已经包含了理解整个框架的最小路径Agent对象承载模型配置和系统提示词run方法接收用户输入返回运行结果。后面的类型安全实现都是在这个骨架上加肉。3.2 用类型安全模型约束Agent的输出接下来进入核心环节让Agent的输出不再是自由文本而是一个有契约的结构化对象。假设场景是工单分诊Agent输入一段用户报障文本输出分类、紧急度、置信度和一句话摘要。我先定义一个Pydantic模型from typing import Literal from pydantic import BaseModel, Field class TicketAssessment(BaseModel): category: Literal[网络故障, 账号安全, 计费问题, 其他] Field(description工单分类) urgency: Literal[低, 中, 高, 紧急] Field(description紧急程度) confidence: float Field(ge0.0, le1.0, description本次判断置信度) summary: str Field(max_length100, description一句话摘要)如果你的Python版本低于3.10Literal要从typing_extensions导入Pydantic本身也会用到这个类型。定义好模型后把它传给Agent的result_type参数agent Agent( your-model-identifier, system_prompt( 你是工单分诊助手。请根据用户描述判断工单分类和紧急程度。 分类必须在给定枚举中紧急程度也必须在给定枚举中 置信度是0到1之间的浮点数不要输出额外解释。 ), result_typeTicketAssessment, max_retries3, )运行后result.output就是一个TicketAssessment实例字段和类型都经过校验可以直接被下游逻辑消费。一旦模型返回的JSON解析失败或字段校验失败框架会按max_retries设定的次数让模型重新生成。这带来的改变是质变级别的。你写的代码不再假设模型“大概率会守规矩”而是要求它“必须守规矩不守就重来”。只要重试仍然失败异常会在你的进程里抛出你可以捕获并做降级处理这比拿到一份坏数据然后到处排查要舒服得多。3.3 工具注册与依赖注入的完整案例工单分诊不能只靠模型猜需要调用真实数据辅助判断。比如用户说“我的账号在异地登录了”Agent应该能查询历史登录记录判断是否真的存在风险。先定义依赖类型。依赖是每个请求独立的所以我把当前请求ID也放在里面from dataclasses import dataclass dataclass class AgentDeps: security_service: object # 安全服务客户端真实项目替换为具体类型 ticket_repo: object # 工单仓库 request_id: str # 本次请求ID再定义Agent并在构造参数里声明依赖类型agent Agent( your-model-identifier, deps_typeAgentDeps, result_typeTicketAssessment, system_prompt你是工单分诊助手..., )然后通过装饰器注册工具。工具函数注意它的第一参数是上下文对象框架会把运行时的依赖注入进来from pydantic_ai import RunContext agent.tool async def query_login_logs( ctx: RunContext[AgentDeps], account: str, days: int 30 ) - list[dict]: 查询指定账号最近N天的登录日志用于判断账号是否异常。 return await ctx.deps.security_service.get_login_logs(account, days)工具函数有几个细节值得专门说。第一个参数RunContext必须放在最前面这是框架的约定模型生成的参数会从第二个参数开始映射。函数签名里的注释很重要模型会读它来理解工具的用途和参数含义写得越清楚调用成功率越高。返回类型要尽量具体框架会根据返回类型对工具输出做解释如果返回裸字符串模型后续推理就容易失去抓手。运行Agent时把依赖对象传进去async def main(): deps AgentDeps( security_serviceSomeService(), ticket_repoSomeRepo(), request_idreq-001, ) result await agent.run( 我的账号昨晚在异地登录了请帮我评估风险, depsdeps, ) print(result.output)这样Agent在处理过程中可以自主决定是否调用查询工具再把工具结果作为上下文综合判断最后输出结构化工单评估。整个调用链路里工具返回值是结构化的最终输出也是结构化的没有一处需要靠正则去捞信息。3.4 让Agent区分轻重缓急紧急度评估场景实现上面已经是一个能用的分诊Agent我们再展开一个更完整的场景看Agent如何把工具调用和结构化输出串起来。业务规则是账号安全问题优先级最高其次计费问题网络波动类问题往往不是紧急级。但规则不能写死因为用户描述可能是“网连不上了但用手机热点还能上”模型需要结合语义判断。同时模型可以调用“查询账号状态”工具拿到账号是否正常的信息来辅助判断。这是我理解的生产级Agent典型特征业务规则不可能穷尽所有输入必须靠模型理解语义但最终决策必须落到确定性的枚举和字段上。代码层面大致如下class AccountStatus(BaseModel): status: Literal[正常, 已被锁定, 存在风险] last_active_days: int failed_attempts: int agent.tool async def check_account_status( ctx: RunContext[AgentDeps], account_id: str ) - AccountStatus: 查询账号当前状态返回状态、最近活跃天数和失败尝试次数。 raw await ctx.deps.security_service.check(account_id) return AccountStatus(**raw) agent.tool async def get_incident_history( ctx: RunContext[AgentDeps], account_id: str ) - list[dict]: 查询该账号历史工单记录用于识别重复问题。 return await ctx.deps.ticket_repo.list_by_account(account_id)Agent运行时会先分析用户描述决定是否需要调用工具。比如用户提到“账号被锁定”它会去查AccountStatus确认状态如果用户只说“网慢”它可能不调用工具直接分诊。这个“决定调不调工具”的动作也由模型自己完成只不过它的选择空间被工具Schema限制住了不会调用不存在的函数。系统提示词我也写出来供参考你是企业级工单分诊Agent。 工作步骤 1. 根据用户报障文本判断是否需要查询账号状态或历史工单 2. 必要时调用工具补充事实 3. 结合工具返回结果输出最终评估。 输出要求 - category只能是网络故障、账号安全、计费问题、其他 - urgency只能是低、中、高、紧急 - confidence是0到1浮点数 - summary不超过100字。这种“步骤输出约束”的提示词和结果模型配合实测下来比单纯丢一句“请规范输出”稳定得多。因为提示词给了决策链路结果模型给了最终格式模型在每一步都知道自己在干什么而不是靠猜。3.5 可观测性日志、指标与重试策略生产级应用如果不知道模型每次想什么、花了多少钱、重试了几次那基本就是盲人骑瞎马。PydanticAI的运行结果对象里含有很多可观测信息但我在项目里还会包一层自己的统一入口函数来做日志。每个Agent调用都会记录这些信息请求ID、Agent名称、模型标识、输入文本长度、调用耗时、输出对象、重试次数和异常信息。下面是一段接近实践的伪代码import logging import time logger logging.getLogger(agent) async def run_with_logging(agent, prompt, deps): start time.monotonic() try: result await agent.run(prompt, depsdeps) elapsed time.monotonic() - start output result.output logger.info( { event: agent_completed, request_id: deps.request_id, agent: getattr(agent, name, unknown), elapsed_ms: round(elapsed * 1000, 2), output: ( output.model_dump() if hasattr(output, model_dump) else str(output) ), } ) return result except Exception as exc: logger.error( { event: agent_failed, request_id: deps.request_id, error: str(exc), } ) raise重试策略要单独说。max_retries参数虽然好用但模型重试一次就是一次真实调用和真实费用。我通常把max_retries控制在2到3次既不轻易放过一次侥幸成功也不至于无限烧钱。对所有验证失败的请求记录失败原因按原因归类统计这些数据要反哺给提示词和结果模型的优化而不是只停留在日志里。3.6 测试策略从模拟响应到影子回放生产级系统得能测试。我分三个层级来测Agent每个层级都不依赖真实模型调用所以测试又快又稳。第一层是契约测试。不调模型直接用手写数据测试Pydantic模型的校验逻辑确保枚举、范围、长度这些约束符合预期。这一层覆盖的是最坏情况比如模型返回了超长的summary、负数置信度、枚举外的分类看模型对象能不能把它们挡在门外。第二层是集成测试。把工具函数替换成内存实现用一段固定的模拟模型响应跑通Agent的完整链路。做法是给Agent注入一组预先准备好的消息或者在工具层做好mock。因为输出是结构化对象断言非常干脆一句assert result.output.category 账号安全就能完成验证不需要对一大段文本做模糊匹配。第三层是回归测试。把线上真实的用户输入和模型输出录制下来在提示词或代码改动后回放比对。这能有效发现“改了一句话导致整个输出风格变了”的问题。我在一次提示词微调后就是靠回归测试发现某个分类的措辞变了影响了下游的工单分组逻辑。这三层测试要写进CI里。因为类型安全框架的存在回放比对不需要复杂的语义相似度算法直接对比模型输出对象的字段值即可工程成本比我预想中低很多。4. 常见问题与排查技巧实录4.1 输出字段漂移与幻字段处理现象结果模型定义了四个字段模型却返回了五个字段多出来的字段叫recommendation。大多数情况下Pydantic校验时会忽略多余字段但如果后续版本启用严格模式就会直接报错。处理方式要先问业务立场多余字段允许吗如果不允许把Pydantic模型配置为严格模式同时提示词里写明“不要输出任何额外字段”如果允许就在模型里显式声明一个可选的extra字段用于兜底避免框架直接报错。我个人的习惯是审计期允许extra字段存在。模型确实会脑补你直接报错会导致重试次数暴涨线上费用跟着涨。等输出稳定后再开启严格模式反而更安全。这个先后顺序比“一上来就追求最严”要务实得多。4.2 工具参数解析失败的三种典型场景最常见的问题就是Agent要调用工具但模型生成的参数过不了校验。典型场景有三个。第一种工具参数名含义模糊。比如参数叫data模型根本不知道要传什么。解决办法是把参数名改成行为指向明确的词并在文档字符串里写清格式和示例。第二种嵌套类型过深。模型生成嵌套JSON时容易出错比如List里套dictdict里又套了多层。解决办法是工具参数尽量扁平化能不嵌套就不嵌套。第三种必填参数太多模型漏填。解决办法是把有默认值的参数全部赋默认值只保留真正必填的参数减少模型的负担。排查这类问题最有效的方式是看框架抛出的校验错误信息它会明确指出哪个字段校验失败。不要急着改提示词先看是不是工具签名设计得不合理。我见过太多人把锅甩给模型实际上工具函数签名本身就是逼着模型犯错。4.3 重试风暴与成本失控现象某天晚上突然收到告警Agent请求量没变但模型调用费用翻了三倍。排查后发现是模型输出连续校验失败每个请求都自动重试了三次。这就是max_retries设置过大的典型事故。我建议生产环境max_retries不超过3。对所有重试事件打指标当某个Agent的重试率超过阈值就告警。如果重试率长时间很高大概率是结果模型约束和实际业务不匹配要重新设计结果模型而不是靠重试硬扛。重试解决的是偶发性的格式问题解决不了系统性的输出偏移。另外像ModelRetry这类主动要求模型重新生成的机制也要谨慎使用。能用一次判断解决的问题别让它转两圈。每次主动重试都会增加模型总输出tokentoken就是成本。4.4 依赖状态共享与并发安全现象上线后偶尔出现A用户的请求拿到了B用户的数据。排查代码后发现工具函数里直接用了模块级全局变量保存当前账号多并发请求时产生了串线。生产级Agent跑在并发环境里依赖注入必须遵循“请求级数据走请求级依赖”这个原则。全局单例只放线程安全的连接、配置和日志客户端。每个请求的账号、上下文、业务状态一定要通过上下文对象挂在当前请求的依赖里。我一般还会在依赖对象里放一个request_id所有日志和外部调用都带它。这样问题一旦出现很快能把整条调用链串起来定位到是哪一个Agent、哪一个工具、哪一个用户的请求出了异常。4.5 模型切换带来的兼容性问题现象测试环境用一个小模型跑得好好的切到线上大模型后工具调用格式反而不稳定。原因是不同模型对工具调用的实现和JSON模式的遵循程度不同。处理方式是把模型标识收敛到一个配置组件不要散落在代码里。切换模型时先跑一遍契约测试和回归测试观察两个关键指标工具调用成功率和输出校验通过率。如果大模型反而不如小模型稳定别迷信“越贵越强”。要针对性地调整提示词或模型参数甚至换回原来的模型线上稳才是真的稳。我把这类高频问题整理成一张速查表排查时直接照着看症状可能原因处理方式输出多字段或少字段模型没遵循提示词校验策略过严严格模式提示词兜底或允许extra字段工具参数解析失败工具签名复杂、参数说明不清扁平化参数、补文档示例、减少必填重试次数飙升结果模型约束与实际输出差距大降低max_retries重新设计结果模型并发数据串线用全局变量存请求级状态依赖注入请求数据放上下文对象切换模型后稳定性下降模型工具调用能力差异统一配置模型标识回归对比工具成功率5. 从Demo到生产部署演进的个人经验5.1 配置治理与密钥管理我见过不少项目把模型密钥直接写在代码里或者把模型名散落在各个模块中。生产环境里配置必须集中管理。我会用一个配置组件统一读取环境变量或配置中心至少保证三件事密钥和凭据不进代码仓库模型标识、温度参数、max_retries这些运行时参数可调开发、测试、生产每个环境有独立的配置。配置这个东西看起来不起眼出事就是大事。有一次排查线上问题发现测试环境的模型标识被误用到了生产环境数据全乱了。从那之后我就在配置组件里加了一键校验环境标识的逻辑环境不一致直接拒绝启动。5.2 版本化提示词与模型灰度系统提示词是业务逻辑的一部分应该版本化。我习惯把每个Agent的系统提示词单独存成文件或配置项带上版本号。改动后不是直接全量上线而是先在新版本Agent上跑一批影子请求对比新旧输出确认没有劣化再灰度放量。模型切换也一样。先小流量观察工具调用成功率和输出校验通过率再逐步放量。这条经验是从一次事故里换来的某次升级提示词后没有做灰度结果所有Agent的输出风格突变下游解析代码全部报警只能连夜回滚。自那以后灰度就成了我所有Agent项目的默认动作省不掉。5.3 最后一点私货建议文章写到这里方法论和代码都展示了。最后分享我在实际项目里的体会不要把Agent当成魔法要把它当成一个“会调用工具的异步函数”。一旦你用这个视角看问题很多决策就清晰了。该用类型安全输出就用该加依赖注入就加该上日志监控就上。模型本身会进步但工程边界不会骗你。PydanticAI这套东西最大的价值不是帮你写Prompt而是逼着你把LLM的不可控输出牢牢锁在类型系统的笼子里。踩过几次坑之后我现在新起一个Agent项目第一件事永远是先定义结果模型和依赖类型再写提示词。顺序反了后面基本都要返工。这个习惯可以说是整个项目五里我认为最有普适价值的一条经验了。