前言上一篇已经讲了 Pydantic AI 的基础用法Agent、deps、RunContext、工具调用、结构化输出和输出验证器。这一篇继续往生产场景推进用 Pydantic AI 写一个电商客服 Agent并讲清它上线到 FastAPI 服务时要注意什么。本文主要解决四个问题如何把依赖注入、工具调用、结构化输出组合成一个真实业务 Agent多用户并发时Agent、会话历史、用户身份如何隔离如何用 Logfire 观察一次 Agent 调用里发生了什么Pydantic AI、LangChain、LangGraph 到底怎么选参考示例目录/Users/lanny/Code/labs/agno_demo/learn_pydantic_ai对应文件customer_service.py server.py server_monitored.py logfire_setup.py背景或问题写一个大模型 Demo 很容易resultagent.run_sync(帮我查订单 A001)print(result.output)但真实客服系统至少要面对这些问题用户说“帮我查订单”模型不能凭空编要先查业务数据。用户可能申请退款退款金额超过权限要走审批。回复里不能暴露手机号等隐私信息。前端希望拿到结构化结果例如分类、情绪、回复正文。多个用户同时访问时对话历史不能串。线上排查时要知道模型调用了哪个工具、参数是什么、耗时多久。这正是 Pydantic AI 比较适合的地方它能把“模型推理”和“后端业务规则”分开。模型负责理解意图、选择工具、组织语言后端函数负责查数据、校验权限、执行规则、记录审计。核心思路客服 Agent 可以拆成五层用户消息 - Agent 指令 - deps 注入业务上下文 - tools 查询客户 / 订单 / 退款 - output_type 返回结构化客服结果 - output_validator 拦截隐私或违规输出代码示例下面用一个“电商客服 Agent”串起来。目标用户问订单状态或退款问题时Agent 先查数据库再返回结构化客服回复并拦截隐私信息。第一步模拟数据库_FAKE_CUSTOMERS{1001:{name:张三,vip:True,phone:138****1234},1002:{name:李四,vip:False,phone:139****5678},}_FAKE_ORDERS{A001:{customer_id:1001,status:已发货,amount:299.0},A002:{customer_id:1002,status:待付款,amount:59.0},A003:{customer_id:1001,status:已完成,amount:1299.0},}真实项目里这里可以换成 PostgreSQL、Redis 或内部订单 API。第二步定义依赖fromdataclassesimportdataclass,fieldfromtypingimportAnydataclassclassCustomerServiceDeps:db:dict[str,Any]field(default_factorylambda:{customers:_FAKE_CUSTOMERS,orders:_FAKE_ORDERS,})agent_name:str小智require_approval_above:float500.0这个 deps 里放了三类运行时信息db业务数据源。agent_name当前客服坐席。require_approval_above退款审批阈值。这样 Agent 不需要写死数据库和权限规则运行时传入即可。第三步定义结构化输出frompydanticimportBaseModel,FieldclassServiceResponse(BaseModel):category:strField(description问题分类售前咨询 / 订单查询 / 投诉 / 退款 / 其他)sentiment:strField(description用户情绪正面 / 中性 / 负面)reply:strField(description给用户的正式回复用中文语气友好专业)前端拿到这个结构后可以根据category渲染不同 UI根据sentiment判断是否升级人工。第四步创建客服 Agentfrompydantic_aiimportAgent,RunContext,ModelRetry support_agentAgent(get_model(),deps_typeCustomerServiceDeps,output_typeServiceResponse,system_prompt(你是一个专业的电商客服。规则\n1) 先调用工具查清客户和订单信息再作答不要凭空猜测。\n2) 不要在回复里输出客户的手机号等隐私信息。\n3) 涉及退款时金额超过坐席权限要走审批流程。\n4) 回复必须结构化输出。),)这里的重点是业务约束进入系统提示词结构约束进入output_type运行时依赖进入deps_type。第五步动态系统提示词support_agent.system_promptdefdynamic_agent_prompt(ctx:RunContext[CustomerServiceDeps])-str:depsctx.depsreturnf当前由客服坐席「{deps.agent_name}」负责接待回复开头可以自称。同一个 Agent 可以给不同坐席使用只需要换 deps。第六步注册业务工具support_agent.tooldefget_customer_info(ctx:RunContext[CustomerServiceDeps],customer_id:int)-str:根据客户 ID 查询客户信息。customersctx.deps.db[customers]customercustomers.get(customer_id)ifnotcustomer:returnf未找到客户 ID{customer_id}returnf客户{customer[name]}VIP{是ifcustomer[vip]else否}support_agent.tooldefget_order_status(ctx:RunContext[CustomerServiceDeps],order_id:str)-str:根据订单号查询订单状态。ordersctx.deps.db[orders]orderorders.get(order_id)ifnotorder:returnf未找到订单{order_id}returnf订单{order_id}状态{order[status]}金额{order[amount]}元support_agent.tooldefrequest_refund(ctx:RunContext[CustomerServiceDeps],order_id:str,reason:str)-str:为某笔订单发起退款申请。depsctx.deps orderdeps.db[orders].get(order_id)ifnotorder:returnf未找到订单{order_id}无法退款amountfloat(order[amount])ifamountdeps.require_approval_above:returnf订单{order_id}金额{amount}元超过坐席权限已提交人工审批。returnf订单{order_id}退款已受理原因{reason}注意模型只负责决定“是否调用工具、用什么参数调用工具”。真正查数据、判断权限、执行退款逻辑的仍然是后端函数。第七步输出验证器support_agent.output_validatordefcheck_no_sensitive_info(ctx:RunContext[CustomerServiceDeps],value:ServiceResponse,)-ServiceResponse:replyvalue.reply digits.join(chforchinreplyifch.isdigit())iflen(digits)11:raiseModelRetry(回复中疑似包含手机号等隐私信息请去掉具体号码后再回复。)returnvalue它的职责是即使模型结构化输出成功也不能把疑似手机号这类隐私写进最终回复。第八步封装业务入口defrun_support(user_message:str,deps:CustomerServiceDeps|NoneNone,)-ServiceResponse:ifdepsisNone:depsCustomerServiceDeps()resultsupport_agent.run_sync(user_message,depsdeps)returnresult.output业务代码只需要调用resprun_support(你好我是客户 1001帮我看看订单 A001 现在什么状态了)print(resp.category)print(resp.sentiment)print(resp.reply)运行结果或效果说明可能得到类似结果分类订单查询 情绪中性 回复您好我是小智。已为您查询到订单 A001目前状态为已发货订单金额为 299 元。请您耐心等待物流更新。如果用户申请大额退款用户订单 A003 我要退款原因是不想要了。 工具返回订单 A003 金额 1299.0 元超过坐席权限已提交人工审批。 最终回复 您好我是小智。您的订单 A003 金额超过当前坐席直接处理权限已为您提交人工审批。后续会有专员继续跟进退款进度。从这个例子可以看到完整闭环prompt 定规则。deps 传上下文。tool 查业务数据。output_type 固定返回结构。output_validator 做最后校验。业务代码拿到强类型对象。FastAPI 部署并发、会话隔离和流式输出脚本跑通之后下一步通常是部署成 HTTP 服务。1. Agent 可以全局复用本地server.py里的核心原则是Agent 对象无状态创建一次全局复用。不要每个请求都 new 一个 Agent。推荐在模块级定义chat_agentAgent(get_model(),deps_typeUserDeps,output_typeChatOutput,system_prompt你是一个友好的中文助手回复简洁必须以结构化数据返回。,)用户相关状态不要放在 Agent 实例里而是放在deps当前请求用户身份、权限、VIP 状态。message_history当前会话历史。外部存储Redis、数据库或内存 store。2. FastAPI 里使用 await agent.run在 Web 服务里不要用run_sync()应该用异步 APIresultawaitchat_agent.run(req.message,depsdeps,message_historyhistory,)原因是 FastAPI 本身运行在事件循环里run_sync()可能触发RuntimeError: This event loop is already running3. 用 session_id 隔离对话历史一个常见流程是请求进入 - 根据 session_id 读取历史 - 根据 user_id 构造 deps - await agent.run(..., message_historyhistory) - result.all_messages() 写回 store伪代码app.post(/chat/{session_id})asyncdefchat(session_id:str,req:ChatRequest,user_id:str,user_name:str):historyawaitstore.get_history(session_id)depsUserDeps(user_iduser_id,user_nameuser_name)resultawaitchat_agent.run(req.message,depsdeps,message_historyhistory,)awaitstore.save_history(session_id,result.all_messages())returnresult.output开发环境可以用内存 dict生产环境建议换 Redis 或数据库。多 worker、多机器时进程内存不共享不能依赖内存保存会话。4. 用 Semaphore 控制 LLM 并发你的 FastAPI 服务可能能接住大量请求但模型供应商通常有并发、RPM、TPM 限制。可以用asyncio.Semaphore包住 LLM 调用_LLM_SEMAPHOREasyncio.Semaphore(20)asyncwith_LLM_SEMAPHORE:resultawaitchat_agent.run(...)这样可以避免高峰期同时打出太多模型请求触发 429 或供应商限流。5. 流式输出用 SSE聊天产品常需要打字机效果可以用run_stream()StreamingResponsefromfastapi.responsesimportStreamingResponseapp.post(/chat/{session_id}/stream)asyncdefchat_stream(session_id:str,req:ChatRequest)-StreamingResponse:asyncdefevent_generator():historyawaitstore.get_history(session_id)asyncwithchat_agent.run_stream(req.message,message_historyhistory,)asrun:asyncfortextinrun.stream_text():yieldfdata:{text}\n\nfinal_textawaitrun.get_output()awaitstore.save_history(session_id,run.all_messages())yieldfdata:{final_text}\n\nreturnStreamingResponse(event_generator(),media_typetext/event-stream)Logfire 监控看清 Agent 内部发生了什么Agent 出问题时可能不是模型本身的问题而是prompt 写得不清楚。工具描述不够明确。工具参数提取错了。历史消息太长或污染了。输出结构校验失败。供应商接口慢或失败。所以可观测性很重要。Pydantic AI 和 Logfire 同属 Pydantic 生态集成比较自然。本地示例封装了一个setup_monitoring()importlogfiredefsetup_monitoring(service_name:strpydantic-ai-server)-None:logfire.configure(service_nameservice_name,send_to_logfireFalse,)logfire.instrument_pydantic_ai()logfire.instrument_httpx(capture_allTrue)logfire.instrument_system_metrics()FastAPI 也可以埋点definstrument_fastapi_app(app:Any)-None:logfire.instrument_fastapi(appapp)开启后一次请求里可以看到HTTP 请求。Agent run。模型请求和响应。工具调用参数和返回值。Token 用量。每一步耗时。自定义业务字段例如user_id、session_id。生产环境要注意capture_allTrue可能记录完整 prompt、用户输入和模型输出。真实业务里要结合脱敏、采样和权限控制使用。与 LangChain、LangGraph 的对比Pydantic AI、LangChain、LangGraph 经常被放在一起比较但它们解决的问题并不完全一样。维度Pydantic AILangChainLangGraph核心定位类型安全的 Python Agent 框架LLM 应用开发组件生态可控、持久、状态化的 Agent 工作流编排上手感觉像 FastAPI Pydantic像一套 LLM 工具箱像用图结构写状态机最突出优势结构化输出、依赖注入、类型提示、校验体验好集成多、资料多、生态成熟状态管理、分支循环、人机协作、长流程控制强适合场景中小型 Agent、结构化提取、工具调用、后端服务集成RAG、链式调用、模型/向量库/工具快速接入多步骤 Agent、审批流、可恢复执行、复杂编排复杂度较低中等较高类型安全很强Pydantic 是核心设计取决于具体写法支持类型化状态但关注点在图运行工具调用函数签名 docstring RunContextTool/Toolkit 生态丰富通常作为图节点或 Agent 节点的一部分多轮记忆显式传message_history外部存储自己管理Memory 组件和 LangGraph 结合更常见状态就是一等公民生产观测Logfire 集成顺手LangSmith 生态成熟LangSmith 图执行轨迹更适合复杂流程一句话区分Pydantic AI适合“我要用 Python 写一个类型安全的 Agent 服务”。LangChain适合“我要快速接各种模型、向量库、Retriever、工具和 RAG 组件”。LangGraph适合“我要把 Agent 做成可控流程有状态、有分支、有循环、可中断、可恢复”。什么时候优先选 Pydantic AI如果你的项目符合这些特征Pydantic AI 很合适后端是 Python。已经大量使用 Pydantic / FastAPI。核心痛点是结构化输出、工具调用、类型校验。Agent 流程不算特别复杂。希望代码像普通后端服务一样可读、可测、可维护。典型例子智能客服。表单自动填写。工单分类。简历解析。合同字段提取。数据分析助手。企业内部工具助手。什么时候优先选 LangChain如果你的重点是快速接入生态LangChain 仍然很有优势。例如需要连接很多向量数据库。需要很多现成 document loader。需要快速搭 RAG pipeline。团队已有 LangChain 代码资产。希望复用社区里大量教程和模板。LangChain 的强项是“组件生态”。它能帮你快速把模型、检索、工具、文档加载器、向量库接起来。什么时候优先选 LangGraph当你的 Agent 开始出现这些需求就该认真考虑 LangGraph多步骤任务不能一次调用结束。流程里有明确状态例如待检索、待审批、待执行、待人工确认。需要循环例如“生成代码 - 运行测试 - 修复错误 - 再测”。需要中断和恢复例如人工审批后继续执行。需要把复杂 Agent 拆成多个节点每个节点独立控制。LangGraph 更像 Agent 工作流运行时不只是“调用大模型的库”。它们不是互斥关系真实项目里这几个工具并不是非此即彼。一个合理组合可能是Pydantic AI 负责单个类型安全 Agent LangGraph 负责多步骤状态流转和人工审批 LangChain 负责接入部分 RAG 生态组件也就是说Pydantic AI 更适合作为“Agent 节点的工程实现”LangGraph 更适合作为“多个节点之间的流程控制”。常见问题与避坑1. Agent 无状态不代表应用无状态Agent 可以全局复用但用户历史、用户身份、权限、业务上下文必须外部管理。推荐结构Agent 全局复用 deps 按请求构造 message_history 按 session_id 读取 业务数据从数据库或 API 获取2. 内存会话存储只能用于开发单进程开发时用内存 dict 很方便。但多 worker 或多机器部署时每个进程都有自己的内存。用户第一次请求打到 worker A第二次请求打到 worker B历史就可能丢失。生产环境建议用 Redis、PostgreSQL 或 MongoDB。3. 高风险工具必须后端强校验退款、删除、支付、发券、修改权限这类动作不能让模型直接决定。推荐做法工具函数里做权限校验。超阈值操作走审批。操作记录审计日志。必要时加入人工确认。4. 监控里不要无脑记录敏感信息Logfire 的完整请求响应记录对调试很有帮助但也可能记录隐私数据。生产环境建议对手机号、身份证、邮箱等字段脱敏。对 trace 做采样。区分开发和生产配置。控制观测平台访问权限。总结Pydantic AI 很适合把单个 Agent 做成可维护的 Python 后端模块deps 管上下文 tool 管业务能力 output_type 管输出结构 output_validator 管业务兜底 message_history 管会话 FastAPI 管服务入口 Logfire 管可观测性如果你的目标是写一个客服、工单、表单提取、内部助手这类 Agent 服务Pydantic AI 的开发体验会很顺手。如果你需要大量 RAG 生态组件LangChain 更有优势。如果你要做复杂多节点流程、人工审批、可中断恢复LangGraph 更适合。三者不是非此即彼关键是把它们放在正确的位置上。参考资料Pydantic AI 官方文档Pydantic AI Agents 文档Pydantic AI Dependencies 文档Pydantic AI Output 文档LangChain Python 官方文档LangGraph 官方文档
Pydantic AI 入门（二）：客服 Agent 实战、FastAPI 部署与框架选型
前言上一篇已经讲了 Pydantic AI 的基础用法Agent、deps、RunContext、工具调用、结构化输出和输出验证器。这一篇继续往生产场景推进用 Pydantic AI 写一个电商客服 Agent并讲清它上线到 FastAPI 服务时要注意什么。本文主要解决四个问题如何把依赖注入、工具调用、结构化输出组合成一个真实业务 Agent多用户并发时Agent、会话历史、用户身份如何隔离如何用 Logfire 观察一次 Agent 调用里发生了什么Pydantic AI、LangChain、LangGraph 到底怎么选参考示例目录/Users/lanny/Code/labs/agno_demo/learn_pydantic_ai对应文件customer_service.py server.py server_monitored.py logfire_setup.py背景或问题写一个大模型 Demo 很容易resultagent.run_sync(帮我查订单 A001)print(result.output)但真实客服系统至少要面对这些问题用户说“帮我查订单”模型不能凭空编要先查业务数据。用户可能申请退款退款金额超过权限要走审批。回复里不能暴露手机号等隐私信息。前端希望拿到结构化结果例如分类、情绪、回复正文。多个用户同时访问时对话历史不能串。线上排查时要知道模型调用了哪个工具、参数是什么、耗时多久。这正是 Pydantic AI 比较适合的地方它能把“模型推理”和“后端业务规则”分开。模型负责理解意图、选择工具、组织语言后端函数负责查数据、校验权限、执行规则、记录审计。核心思路客服 Agent 可以拆成五层用户消息 - Agent 指令 - deps 注入业务上下文 - tools 查询客户 / 订单 / 退款 - output_type 返回结构化客服结果 - output_validator 拦截隐私或违规输出代码示例下面用一个“电商客服 Agent”串起来。目标用户问订单状态或退款问题时Agent 先查数据库再返回结构化客服回复并拦截隐私信息。第一步模拟数据库_FAKE_CUSTOMERS{1001:{name:张三,vip:True,phone:138****1234},1002:{name:李四,vip:False,phone:139****5678},}_FAKE_ORDERS{A001:{customer_id:1001,status:已发货,amount:299.0},A002:{customer_id:1002,status:待付款,amount:59.0},A003:{customer_id:1001,status:已完成,amount:1299.0},}真实项目里这里可以换成 PostgreSQL、Redis 或内部订单 API。第二步定义依赖fromdataclassesimportdataclass,fieldfromtypingimportAnydataclassclassCustomerServiceDeps:db:dict[str,Any]field(default_factorylambda:{customers:_FAKE_CUSTOMERS,orders:_FAKE_ORDERS,})agent_name:str小智require_approval_above:float500.0这个 deps 里放了三类运行时信息db业务数据源。agent_name当前客服坐席。require_approval_above退款审批阈值。这样 Agent 不需要写死数据库和权限规则运行时传入即可。第三步定义结构化输出frompydanticimportBaseModel,FieldclassServiceResponse(BaseModel):category:strField(description问题分类售前咨询 / 订单查询 / 投诉 / 退款 / 其他)sentiment:strField(description用户情绪正面 / 中性 / 负面)reply:strField(description给用户的正式回复用中文语气友好专业)前端拿到这个结构后可以根据category渲染不同 UI根据sentiment判断是否升级人工。第四步创建客服 Agentfrompydantic_aiimportAgent,RunContext,ModelRetry support_agentAgent(get_model(),deps_typeCustomerServiceDeps,output_typeServiceResponse,system_prompt(你是一个专业的电商客服。规则\n1) 先调用工具查清客户和订单信息再作答不要凭空猜测。\n2) 不要在回复里输出客户的手机号等隐私信息。\n3) 涉及退款时金额超过坐席权限要走审批流程。\n4) 回复必须结构化输出。),)这里的重点是业务约束进入系统提示词结构约束进入output_type运行时依赖进入deps_type。第五步动态系统提示词support_agent.system_promptdefdynamic_agent_prompt(ctx:RunContext[CustomerServiceDeps])-str:depsctx.depsreturnf当前由客服坐席「{deps.agent_name}」负责接待回复开头可以自称。同一个 Agent 可以给不同坐席使用只需要换 deps。第六步注册业务工具support_agent.tooldefget_customer_info(ctx:RunContext[CustomerServiceDeps],customer_id:int)-str:根据客户 ID 查询客户信息。customersctx.deps.db[customers]customercustomers.get(customer_id)ifnotcustomer:returnf未找到客户 ID{customer_id}returnf客户{customer[name]}VIP{是ifcustomer[vip]else否}support_agent.tooldefget_order_status(ctx:RunContext[CustomerServiceDeps],order_id:str)-str:根据订单号查询订单状态。ordersctx.deps.db[orders]orderorders.get(order_id)ifnotorder:returnf未找到订单{order_id}returnf订单{order_id}状态{order[status]}金额{order[amount]}元support_agent.tooldefrequest_refund(ctx:RunContext[CustomerServiceDeps],order_id:str,reason:str)-str:为某笔订单发起退款申请。depsctx.deps orderdeps.db[orders].get(order_id)ifnotorder:returnf未找到订单{order_id}无法退款amountfloat(order[amount])ifamountdeps.require_approval_above:returnf订单{order_id}金额{amount}元超过坐席权限已提交人工审批。returnf订单{order_id}退款已受理原因{reason}注意模型只负责决定“是否调用工具、用什么参数调用工具”。真正查数据、判断权限、执行退款逻辑的仍然是后端函数。第七步输出验证器support_agent.output_validatordefcheck_no_sensitive_info(ctx:RunContext[CustomerServiceDeps],value:ServiceResponse,)-ServiceResponse:replyvalue.reply digits.join(chforchinreplyifch.isdigit())iflen(digits)11:raiseModelRetry(回复中疑似包含手机号等隐私信息请去掉具体号码后再回复。)returnvalue它的职责是即使模型结构化输出成功也不能把疑似手机号这类隐私写进最终回复。第八步封装业务入口defrun_support(user_message:str,deps:CustomerServiceDeps|NoneNone,)-ServiceResponse:ifdepsisNone:depsCustomerServiceDeps()resultsupport_agent.run_sync(user_message,depsdeps)returnresult.output业务代码只需要调用resprun_support(你好我是客户 1001帮我看看订单 A001 现在什么状态了)print(resp.category)print(resp.sentiment)print(resp.reply)运行结果或效果说明可能得到类似结果分类订单查询 情绪中性 回复您好我是小智。已为您查询到订单 A001目前状态为已发货订单金额为 299 元。请您耐心等待物流更新。如果用户申请大额退款用户订单 A003 我要退款原因是不想要了。 工具返回订单 A003 金额 1299.0 元超过坐席权限已提交人工审批。 最终回复 您好我是小智。您的订单 A003 金额超过当前坐席直接处理权限已为您提交人工审批。后续会有专员继续跟进退款进度。从这个例子可以看到完整闭环prompt 定规则。deps 传上下文。tool 查业务数据。output_type 固定返回结构。output_validator 做最后校验。业务代码拿到强类型对象。FastAPI 部署并发、会话隔离和流式输出脚本跑通之后下一步通常是部署成 HTTP 服务。1. Agent 可以全局复用本地server.py里的核心原则是Agent 对象无状态创建一次全局复用。不要每个请求都 new 一个 Agent。推荐在模块级定义chat_agentAgent(get_model(),deps_typeUserDeps,output_typeChatOutput,system_prompt你是一个友好的中文助手回复简洁必须以结构化数据返回。,)用户相关状态不要放在 Agent 实例里而是放在deps当前请求用户身份、权限、VIP 状态。message_history当前会话历史。外部存储Redis、数据库或内存 store。2. FastAPI 里使用 await agent.run在 Web 服务里不要用run_sync()应该用异步 APIresultawaitchat_agent.run(req.message,depsdeps,message_historyhistory,)原因是 FastAPI 本身运行在事件循环里run_sync()可能触发RuntimeError: This event loop is already running3. 用 session_id 隔离对话历史一个常见流程是请求进入 - 根据 session_id 读取历史 - 根据 user_id 构造 deps - await agent.run(..., message_historyhistory) - result.all_messages() 写回 store伪代码app.post(/chat/{session_id})asyncdefchat(session_id:str,req:ChatRequest,user_id:str,user_name:str):historyawaitstore.get_history(session_id)depsUserDeps(user_iduser_id,user_nameuser_name)resultawaitchat_agent.run(req.message,depsdeps,message_historyhistory,)awaitstore.save_history(session_id,result.all_messages())returnresult.output开发环境可以用内存 dict生产环境建议换 Redis 或数据库。多 worker、多机器时进程内存不共享不能依赖内存保存会话。4. 用 Semaphore 控制 LLM 并发你的 FastAPI 服务可能能接住大量请求但模型供应商通常有并发、RPM、TPM 限制。可以用asyncio.Semaphore包住 LLM 调用_LLM_SEMAPHOREasyncio.Semaphore(20)asyncwith_LLM_SEMAPHORE:resultawaitchat_agent.run(...)这样可以避免高峰期同时打出太多模型请求触发 429 或供应商限流。5. 流式输出用 SSE聊天产品常需要打字机效果可以用run_stream()StreamingResponsefromfastapi.responsesimportStreamingResponseapp.post(/chat/{session_id}/stream)asyncdefchat_stream(session_id:str,req:ChatRequest)-StreamingResponse:asyncdefevent_generator():historyawaitstore.get_history(session_id)asyncwithchat_agent.run_stream(req.message,message_historyhistory,)asrun:asyncfortextinrun.stream_text():yieldfdata:{text}\n\nfinal_textawaitrun.get_output()awaitstore.save_history(session_id,run.all_messages())yieldfdata:{final_text}\n\nreturnStreamingResponse(event_generator(),media_typetext/event-stream)Logfire 监控看清 Agent 内部发生了什么Agent 出问题时可能不是模型本身的问题而是prompt 写得不清楚。工具描述不够明确。工具参数提取错了。历史消息太长或污染了。输出结构校验失败。供应商接口慢或失败。所以可观测性很重要。Pydantic AI 和 Logfire 同属 Pydantic 生态集成比较自然。本地示例封装了一个setup_monitoring()importlogfiredefsetup_monitoring(service_name:strpydantic-ai-server)-None:logfire.configure(service_nameservice_name,send_to_logfireFalse,)logfire.instrument_pydantic_ai()logfire.instrument_httpx(capture_allTrue)logfire.instrument_system_metrics()FastAPI 也可以埋点definstrument_fastapi_app(app:Any)-None:logfire.instrument_fastapi(appapp)开启后一次请求里可以看到HTTP 请求。Agent run。模型请求和响应。工具调用参数和返回值。Token 用量。每一步耗时。自定义业务字段例如user_id、session_id。生产环境要注意capture_allTrue可能记录完整 prompt、用户输入和模型输出。真实业务里要结合脱敏、采样和权限控制使用。与 LangChain、LangGraph 的对比Pydantic AI、LangChain、LangGraph 经常被放在一起比较但它们解决的问题并不完全一样。维度Pydantic AILangChainLangGraph核心定位类型安全的 Python Agent 框架LLM 应用开发组件生态可控、持久、状态化的 Agent 工作流编排上手感觉像 FastAPI Pydantic像一套 LLM 工具箱像用图结构写状态机最突出优势结构化输出、依赖注入、类型提示、校验体验好集成多、资料多、生态成熟状态管理、分支循环、人机协作、长流程控制强适合场景中小型 Agent、结构化提取、工具调用、后端服务集成RAG、链式调用、模型/向量库/工具快速接入多步骤 Agent、审批流、可恢复执行、复杂编排复杂度较低中等较高类型安全很强Pydantic 是核心设计取决于具体写法支持类型化状态但关注点在图运行工具调用函数签名 docstring RunContextTool/Toolkit 生态丰富通常作为图节点或 Agent 节点的一部分多轮记忆显式传message_history外部存储自己管理Memory 组件和 LangGraph 结合更常见状态就是一等公民生产观测Logfire 集成顺手LangSmith 生态成熟LangSmith 图执行轨迹更适合复杂流程一句话区分Pydantic AI适合“我要用 Python 写一个类型安全的 Agent 服务”。LangChain适合“我要快速接各种模型、向量库、Retriever、工具和 RAG 组件”。LangGraph适合“我要把 Agent 做成可控流程有状态、有分支、有循环、可中断、可恢复”。什么时候优先选 Pydantic AI如果你的项目符合这些特征Pydantic AI 很合适后端是 Python。已经大量使用 Pydantic / FastAPI。核心痛点是结构化输出、工具调用、类型校验。Agent 流程不算特别复杂。希望代码像普通后端服务一样可读、可测、可维护。典型例子智能客服。表单自动填写。工单分类。简历解析。合同字段提取。数据分析助手。企业内部工具助手。什么时候优先选 LangChain如果你的重点是快速接入生态LangChain 仍然很有优势。例如需要连接很多向量数据库。需要很多现成 document loader。需要快速搭 RAG pipeline。团队已有 LangChain 代码资产。希望复用社区里大量教程和模板。LangChain 的强项是“组件生态”。它能帮你快速把模型、检索、工具、文档加载器、向量库接起来。什么时候优先选 LangGraph当你的 Agent 开始出现这些需求就该认真考虑 LangGraph多步骤任务不能一次调用结束。流程里有明确状态例如待检索、待审批、待执行、待人工确认。需要循环例如“生成代码 - 运行测试 - 修复错误 - 再测”。需要中断和恢复例如人工审批后继续执行。需要把复杂 Agent 拆成多个节点每个节点独立控制。LangGraph 更像 Agent 工作流运行时不只是“调用大模型的库”。它们不是互斥关系真实项目里这几个工具并不是非此即彼。一个合理组合可能是Pydantic AI 负责单个类型安全 Agent LangGraph 负责多步骤状态流转和人工审批 LangChain 负责接入部分 RAG 生态组件也就是说Pydantic AI 更适合作为“Agent 节点的工程实现”LangGraph 更适合作为“多个节点之间的流程控制”。常见问题与避坑1. Agent 无状态不代表应用无状态Agent 可以全局复用但用户历史、用户身份、权限、业务上下文必须外部管理。推荐结构Agent 全局复用 deps 按请求构造 message_history 按 session_id 读取 业务数据从数据库或 API 获取2. 内存会话存储只能用于开发单进程开发时用内存 dict 很方便。但多 worker 或多机器部署时每个进程都有自己的内存。用户第一次请求打到 worker A第二次请求打到 worker B历史就可能丢失。生产环境建议用 Redis、PostgreSQL 或 MongoDB。3. 高风险工具必须后端强校验退款、删除、支付、发券、修改权限这类动作不能让模型直接决定。推荐做法工具函数里做权限校验。超阈值操作走审批。操作记录审计日志。必要时加入人工确认。4. 监控里不要无脑记录敏感信息Logfire 的完整请求响应记录对调试很有帮助但也可能记录隐私数据。生产环境建议对手机号、身份证、邮箱等字段脱敏。对 trace 做采样。区分开发和生产配置。控制观测平台访问权限。总结Pydantic AI 很适合把单个 Agent 做成可维护的 Python 后端模块deps 管上下文 tool 管业务能力 output_type 管输出结构 output_validator 管业务兜底 message_history 管会话 FastAPI 管服务入口 Logfire 管可观测性如果你的目标是写一个客服、工单、表单提取、内部助手这类 Agent 服务Pydantic AI 的开发体验会很顺手。如果你需要大量 RAG 生态组件LangChain 更有优势。如果你要做复杂多节点流程、人工审批、可中断恢复LangGraph 更适合。三者不是非此即彼关键是把它们放在正确的位置上。参考资料Pydantic AI 官方文档Pydantic AI Agents 文档Pydantic AI Dependencies 文档Pydantic AI Output 文档LangChain Python 官方文档LangGraph 官方文档
