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 很容易：

result=agent.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,fieldfromtypingimportAny@dataclassclassCustomerServiceDeps:db:dict[str,Any]=field(default_factory=lambda:{"customers":_FAKE_CUSTOMERS,"orders":_FAKE_ORDERS,})agent_name:str="小智"require_approval_above:float=500.0

这个 deps 里放了三类运行时信息：

• db：业务数据源。
• agent_name：当前客服坐席。
• require_approval_above：退款审批阈值。

这样 Agent 不需要写死数据库和权限规则，运行时传入即可。

第三步：定义结构化输出

frompydanticimportBaseModel,FieldclassServiceResponse(BaseModel):category:str=Field(description="问题分类：售前咨询 / 订单查询 / 投诉 / 退款 / 其他")sentiment:str=Field(description="用户情绪：正面 / 中性 / 负面")reply:str=Field(description="给用户的正式回复，用中文，语气友好专业")

前端拿到这个结构后，可以根据category渲染不同 UI，根据sentiment判断是否升级人工。

第四步：创建客服 Agent

frompydantic_aiimportAgent,RunContext,ModelRetry support_agent=Agent(get_model(),deps_type=CustomerServiceDeps,output_type=ServiceResponse,system_prompt=("你是一个专业的电商客服。规则：\n""1) 先调用工具查清客户和订单信息，再作答，不要凭空猜测。\n""2) 不要在回复里输出客户的手机号等隐私信息。\n""3) 涉及退款时，金额超过坐席权限要走审批流程。\n""4) 回复必须结构化输出。"),)

这里的重点是：业务约束进入系统提示词，结构约束进入output_type，运行时依赖进入deps_type。

第五步：动态系统提示词

@support_agent.system_promptdefdynamic_agent_prompt(ctx:RunContext[CustomerServiceDeps])->str:deps=ctx.depsreturnf"当前由客服坐席「{deps.agent_name}」负责接待，回复开头可以自称。"

同一个 Agent 可以给不同坐席使用，只需要换 deps。

第六步：注册业务工具

@support_agent.tooldefget_customer_info(ctx:RunContext[CustomerServiceDeps],customer_id:int)->str:"""根据客户 ID 查询客户信息。"""customers=ctx.deps.db["customers"]customer=customers.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:"""根据订单号查询订单状态。"""orders=ctx.deps.db["orders"]order=orders.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:"""为某笔订单发起退款申请。"""deps=ctx.deps order=deps.db["orders"].get(order_id)ifnotorder:returnf"未找到订单{order_id}，无法退款"amount=float(order["amount"])ifamount>deps.require_approval_above:returnf"订单{order_id}金额{amount}元，超过坐席权限，已提交人工审批。"returnf"订单{order_id}退款已受理，原因：{reason}"

注意：模型只负责决定“是否调用工具、用什么参数调用工具”。真正查数据、判断权限、执行退款逻辑的仍然是后端函数。

第七步：输出验证器

@support_agent.output_validatordefcheck_no_sensitive_info(ctx:RunContext[CustomerServiceDeps],value:ServiceResponse,)->ServiceResponse:reply=value.reply digits="".join(chforchinreplyifch.isdigit())iflen(digits)>=11:raiseModelRetry("回复中疑似包含手机号等隐私信息，请去掉具体号码后再回复。")returnvalue

它的职责是：即使模型结构化输出成功，也不能把疑似手机号这类隐私写进最终回复。

第八步：封装业务入口

defrun_support(user_message:str,deps:CustomerServiceDeps|None=None,)->ServiceResponse:ifdepsisNone:deps=CustomerServiceDeps()result=support_agent.run_sync(user_message,deps=deps)returnresult.output

业务代码只需要调用：

resp=run_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_agent=Agent(get_model(),deps_type=UserDeps,output_type=ChatOutput,system_prompt="你是一个友好的中文助手，回复简洁，必须以结构化数据返回。",)

用户相关状态不要放在 Agent 实例里，而是放在：

• deps：当前请求用户身份、权限、VIP 状态。
• message_history：当前会话历史。
• 外部存储：Redis、数据库或内存 store。

2. FastAPI 里使用 await agent.run

在 Web 服务里不要用run_sync()，应该用异步 API：

result=awaitchat_agent.run(req.message,deps=deps,message_history=history,)

原因是 FastAPI 本身运行在事件循环里，run_sync()可能触发：

RuntimeError: This event loop is already running

3. 用 session_id 隔离对话历史

一个常见流程是：

请求进入 -> 根据 session_id 读取历史 -> 根据 user_id 构造 deps -> await agent.run(..., message_history=history) -> result.all_messages() 写回 store

伪代码：

@app.post("/chat/{session_id}")asyncdefchat(session_id:str,req:ChatRequest,user_id:str,user_name:str):history=awaitstore.get_history(session_id)deps=UserDeps(user_id=user_id,user_name=user_name)result=awaitchat_agent.run(req.message,deps=deps,message_history=history,)awaitstore.save_history(session_id,result.all_messages())returnresult.output

开发环境可以用内存 dict，生产环境建议换 Redis 或数据库。多 worker、多机器时，进程内存不共享，不能依赖内存保存会话。

4. 用 Semaphore 控制 LLM 并发

你的 FastAPI 服务可能能接住大量请求，但模型供应商通常有并发、RPM、TPM 限制。

可以用asyncio.Semaphore包住 LLM 调用：

_LLM_SEMAPHORE=asyncio.Semaphore(20)asyncwith_LLM_SEMAPHORE:result=awaitchat_agent.run(...)

这样可以避免高峰期同时打出太多模型请求，触发 429 或供应商限流。

5. 流式输出用 SSE

聊天产品常需要打字机效果，可以用run_stream()+StreamingResponse：

fromfastapi.responsesimportStreamingResponse@app.post("/chat/{session_id}/stream")asyncdefchat_stream(session_id:str,req:ChatRequest)->StreamingResponse:asyncdefevent_generator():history=awaitstore.get_history(session_id)asyncwithchat_agent.run_stream(req.message,message_history=history,)asrun:asyncfortextinrun.stream_text():yieldf"data:{text}\n\n"final_text=awaitrun.get_output()awaitstore.save_history(session_id,run.all_messages())yieldf"data:{final_text}\n\n"returnStreamingResponse(event_generator(),media_type="text/event-stream")

Logfire 监控：看清 Agent 内部发生了什么

Agent 出问题时，可能不是模型本身的问题，而是：

• prompt 写得不清楚。
• 工具描述不够明确。
• 工具参数提取错了。
• 历史消息太长或污染了。
• 输出结构校验失败。
• 供应商接口慢或失败。

所以可观测性很重要。

Pydantic AI 和 Logfire 同属 Pydantic 生态，集成比较自然。

本地示例封装了一个setup_monitoring()：

importlogfiredefsetup_monitoring(service_name:str="pydantic-ai-server")->None:logfire.configure(service_name=service_name,send_to_logfire=False,)logfire.instrument_pydantic_ai()logfire.instrument_httpx(capture_all=True)logfire.instrument_system_metrics()

FastAPI 也可以埋点：

definstrument_fastapi_app(app:Any)->None:logfire.instrument_fastapi(app=app)

开启后，一次请求里可以看到：

• HTTP 请求。
• Agent run。
• 模型请求和响应。
• 工具调用参数和返回值。
• Token 用量。
• 每一步耗时。
• 自定义业务字段，例如user_id、session_id。

生产环境要注意：capture_all=True可能记录完整 prompt、用户输入和模型输出。真实业务里要结合脱敏、采样和权限控制使用。

与 LangChain、LangGraph 的对比

Pydantic AI、LangChain、LangGraph 经常被放在一起比较，但它们解决的问题并不完全一样。

一句话区分：

• 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 官方文档