1. 项目概述为什么我们需要一个类型安全的 AI Agent最近在捣鼓 AI Agent 开发发现一个挺普遍的问题代码写着写着就成了一团乱麻。特别是当你需要集成多个工具、管理复杂的对话状态还要处理流式输出时那种感觉就像在指挥一支没有经过任何训练的乐队每个乐手模块都在即兴发挥最后出来的声音可想而知。传统的开发方式比如直接用 OpenAI SDK 或者一些早期框架往往把 API 调用、工具调用逻辑、状态管理全部揉在一起缺乏清晰的边界和类型约束。这就导致两个大问题一是调试困难一个参数类型传错了可能要等到运行时才能发现二是代码难以维护和扩展加个新功能就得小心翼翼生怕碰坏了其他地方。这时候PydanticAI进入了我的视野。它不是一个全新的概念而是建立在 Pydantic 这个强大的数据验证库和 Python 的类型提示Type Hints体系之上专门为构建 AI Agent 而生。它的核心卖点就是类型安全Type Safety。简单来说它让你能用写 Python 类和方法的那种清晰、可预测的方式来定义 Agent 的行为、工具的参数和返回结果。编译器或类型检查器如 mypy能在你写代码的时候就帮你揪出很多潜在的错误而不是等到程序跑起来再报一堆让人头疼的异常。这个项目就是带你从零开始用 PydanticAI 搭建一个具备完整功能的 AI Agent。我们会重点攻克几个在实际开发中高频出现的关键环节如何优雅地使用依赖注入来管理 Agent 所需的上下文比如数据库连接、配置信息而不是硬编码在函数里如何规范地注册和使用工具确保工具的参数和返回值都有明确的类型定义最后如何实现流式输出让 Agent 的思考过程或最终答案能够像 ChatGPT 那样一个字一个字地“流”出来提升用户体验。整个过程你会看到类型安全如何让 Agent 的开发从“刀耕火种”走向“精耕细作”。2. 核心设计思路依赖注入、工具与流的三角架构在动手写代码之前我们先得把架构想清楚。一个健壮的 AI Agent尤其是业务逻辑稍复杂的不能把所有东西都塞进一个巨大的main函数里。PydanticAI 鼓励我们采用一种更模块化、更易于测试的设计。我把它总结为“三角架构”三个顶点分别是依赖Dependencies、工具Tools和执行流Execution Streaming。2.1 依赖注入给 Agent 提供“上下文装备”依赖注入听起来高大上其实理念很简单不是让 Agent 自己内部去创建它需要的东西比如数据库客户端、配置对象、其他服务而是由外部“注入”给它。这样做的好处太多了解耦Agent 的逻辑不依赖于具体的实现。今天用 SQLite明天换 PostgreSQL只需要换一个注入的客户端Agent 代码一行不用改。可测试性测试时你可以轻松地注入一个“模拟Mock”的数据库客户端而不是去连接一个真实的数据库。状态管理像用户会话、长期记忆这些状态可以通过依赖来管理和共享。在 PydanticAI 中依赖通常被定义为异步函数async def它们的返回值会被自动提供给 Agent 的run方法或工具函数。例如一个获取当前用户信息的依赖一个获取数据库连接的依赖。2.2 工具注册定义 Agent 的“手脚”工具是 Agent 与外部世界交互的桥梁。查天气、发邮件、读写数据库都需要通过工具。PydanticAI 要求我们用 Pydantic 模型来严格定义工具的输入参数args_schema这直接带来了类型安全。你在代码里调用工具时如果参数类型不对IDE 会直接报错。工具注册的本质是将这些定义好的工具函数“告诉” Agent。PydanticAI 提供了装饰器如agent.tool和显式注册等多种方式。注册后的工具其描述和参数 schema 会被自动编入发送给大语言模型LLM的提示词Prompt中LLM 才知道在什么情况下、以什么格式来调用这个工具。2.3 流式输出让交互过程“活”起来流式输出是现代 AI 应用的标配。它不仅仅是把最终答案一次性返回更重要的是可以实时返回 Agent 的“思考过程”Reasoning或部分结果。这对于需要长时间运行的任务如编写代码、分析长文档至关重要用户不用傻等着能看到进度。PydanticAI 原生支持流式输出。其核心在于stream方法它返回的是一个异步生成器AsyncGenerator。这个生成器不仅会产出最终的答案还会产出中间的关键步骤比如“调用了哪个工具”、“工具返回了什么结果”、“Agent 现在在想什么”。我们需要做的就是处理好这个流将其转化为前端可以消费的格式如 Server-Sent Events。这个三角架构是环环相扣的依赖为工具和 Agent 提供运行环境工具扩展了 Agent 的能力边界流式输出则将依赖和工具协作产生的过程与结果高效地呈现给用户。接下来我们就进入实战环节一步步把它们搭建起来。3. 环境准备与 PydanticAI 核心概念速览工欲善其事必先利其器。我们先来把开发环境搭好并快速理解几个 PydanticAI 的核心模型Model这是后续所有工作的基础。3.1 创建项目与安装依赖我习惯用一个干净的虚拟环境来开始新项目避免包版本冲突。# 创建项目目录并进入 mkdir type-safe-ai-agent cd type-safe-ai-agent # 创建虚拟环境这里用 venv你也可以用 conda python -m venv .venv # 激活虚拟环境 # Windows: .venv\Scripts\activate # Linux/Mac: source .venv/bin/activate接下来安装核心依赖。除了pydantic-ai我们还需要openai作为 LLM 后端当然 PydanticAI 也支持 Anthropic、Gemini 等以及httpx用于可能的网络请求pydantic-settings用来管理配置。pip install pydantic-ai openai httpx pydantic-settings # 可选用于环境变量管理 pip install python-dotenv3.2 理解 PydanticAI 的四大核心模型PydanticAI 的 API 设计围绕着几个核心类展开理解它们的关系至关重要。Agent代理这是最外层的容器代表一个 AI 代理。它绑定了一个具体的 LLM如 GPT-4并包含了一系列的工具、系统提示词System Prompt和运行配置。我们通过实例化Agent类来创建代理。Model模型在 PydanticAI 上下文中Model类是对 LLM 客户端如 OpenAI client的封装。它处理与 LLM API 的实际通信包括格式化请求、解析响应、处理错误等。创建Agent时必须指定一个Model。RunContext运行上下文这是 Agent 单次运行run或stream的上下文环境。它包含了本次运行的所有状态信息比如用户输入messages、注入的依赖结果deps、可用的工具列表等。我们通常不需要直接创建它但会在工具函数和依赖函数中通过参数访问它。Result结果Agent 运行后返回的对象。它包含了最核心的 AI 回复内容data以及完整的运行元数据比如调用了哪些工具tool_calls、消耗的 Token 数usage、本次运行的成本cost等。对于流式输出我们收到的是多个Result对象每个对象代表流中的一个“块”chunk。一个最简单的非流式 Agent 调用流程是这样的你创建Model和Agent然后调用agent.run(“用户问题”)返回一个Result对象从result.data里拿到 AI 的回复。3.3 初始化配置与 Model在实际项目中API Key 等敏感信息肯定不能写死在代码里。我们用pydantic-settings和.env文件来管理。首先在项目根目录创建.env文件OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果你用官方API # 如果使用其他兼容OpenAI的API服务可以修改 BASE_URL # OPENAI_BASE_URLhttps://your.proxy.com/v1然后创建一个config.py文件来定义配置from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str openai_base_url: str https://api.openai.com/v1 # 可以添加其他配置如模型名称 model_name: str gpt-4o-mini class Config: env_file .env settings Settings()接着在main.py或你的应用入口文件中初始化 Modelimport asyncio from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from config import settings # 1. 创建 Model这是与LLM通信的客户端 model OpenAIModel( model_namesettings.model_name, api_keysettings.openai_api_key, base_urlsettings.openai_base_url, ) # 2. 创建 Agent绑定 Model 并设置系统提示词 agent Agent( modelmodel, system_prompt你是一个乐于助人的AI助手。请用中文回答用户的问题。, ) async def main(): # 3. 运行 Agent result await agent.run(你好世界) print(fAI回复: {result.data}) if __name__ __main__: asyncio.run(main())运行这个脚本如果一切配置正确你应该能看到 AI 的问候回复。这标志着你的 PydanticAI 基础环境已经跑通了。接下来我们要为这个 Agent 注入灵魂——依赖和工具。4. 实战依赖注入构建可测试的 Agent 上下文现在我们来给 Agent 加上“依赖”。假设我们要构建一个客服 Agent它需要知道当前咨询的用户是谁用户信息并且能够访问知识库数据库连接。我们不会把这些信息硬编码在工具函数里而是通过依赖注入。4.1 定义依赖函数依赖函数就是普通的async def函数它的参数可以接收RunContext。PydanticAI 会自动调用这些函数并将它们的返回值放入上下文中供后续的 Agent 逻辑或工具使用。我们在项目里创建一个dependencies.py文件from pydantic_ai import RunContext from typing import Dict, Any import httpx # 假设的用户数据库实际项目中可能是Redis、SQL数据库 fake_user_db { user_001: {id: user_001, name: 张三, vip_level: 3}, user_002: {id: user_002, name: 李四, vip_level: 1}, } async def get_current_user(ctx: RunContext) - Dict[str, Any]: 依赖1获取当前用户信息。 在实际Web应用中ctx 里可能包含了从请求头中解析出的用户令牌token。 这里我们模拟从‘ctx’的某个属性比如 ctx.state中获取用户ID。 # 模拟假设我们从上下文的某个地方拿到了用户ID。这里为了演示我们写死一个。 # 真实场景下可能是 user_id ctx.state.get(user_id) user_id user_001 user fake_user_db.get(user_id) if not user: # 如果用户不存在可以返回一个默认用户或抛出错误 # PydanticAI 支持依赖抛出异常这会导致整个 Agent 运行失败 raise ValueError(f用户 {user_id} 不存在) return user async def get_knowledge_base_conn(ctx: RunContext): 依赖2获取知识库连接。 这里我们模拟一个简单的键值存储客户端。实际可能是 Elasticsearch、向量数据库等的连接。 注意这个函数返回的不是数据而是一个“客户端”对象。 # 模拟一个简单的内存知识库 class KnowledgeBaseClient: def __init__(self): self.data { 退货政策: 商品签收后7天内可无理由退货保持商品完好。, 运费说明: 订单满99元包邮不满99元收取10元运费。, 客服时间: 人工客服工作时间周一至周日 9:00-18:00。, } async def query(self, topic: str) - str: return self.data.get(topic, 抱歉未找到相关知识点。) # 每次调用依赖返回一个新的客户端实例或共享的连接池 # 对于数据库连接更常见的做法是注入一个连接池这里简化处理。 return KnowledgeBaseClient() # 可以定义更多依赖如获取系统配置、日志记录器等。4.2 在 Agent 中使用依赖定义了依赖函数后我们需要在创建 Agent 或运行 Agent 时声明它们。有两种主要方式方式一在Agent.run()或Agent.stream()时通过deps参数注入。这种方式灵活适合依赖项在每次运行时可能不同的场景。# 接之前的 main 函数 async def main_with_deps(): user_question 你们的退货政策是怎样的 result await agent.run( user_question, deps[get_current_user, get_knowledge_base_conn] # 传入依赖函数列表 ) print(fAI回复: {result.data}) # 我们还可以从 result 里查看本次运行使用的依赖值调试用 print(f本次运行注入的用户是: {result.deps_results[get_current_user]})方式二将依赖定义为 Agent 的默认依赖。这种方式适合那些全局的、每次运行都需要的依赖比如配置、日志器。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from dependencies import get_knowledge_base_conn from config import settings # 创建 Model (同上) model OpenAIModel(...) # 创建 Agent 时指定默认依赖 agent_with_default_deps Agent( modelmodel, system_prompt你是智能客服助手请根据知识库和用户信息回答问题。, deps[get_knowledge_base_conn], # 默认依赖每次运行都会自动注入 ) async def main_default_deps(): # 这次运行只需要再注入用户依赖即可 result await agent_with_default_deps.run( 我想查询运费。, deps[get_current_user] # 合并默认依赖和本次运行依赖 ) print(result.data)4.3 在工具函数中访问依赖这是依赖注入价值体现的关键点。工具函数可以通过参数直接访问由依赖注入的值。PydanticAI 会自动根据参数名进行匹配。from pydantic_ai import RunContext from pydantic import BaseModel, Field # 首先定义一个工具的输入参数模型 class QueryKnowledgeBaseArgs(BaseModel): topic: str Field(description要查询的知识点主题例如‘退货政策’、‘运费说明’) # 注意工具函数的参数ctx 是 RunContextkb_client 就是依赖 get_knowledge_base_conn 的返回值。 # PydanticAI 会自动将同名的依赖结果传递进来。 async def tool_query_knowledge_base(ctx: RunContext, kb_client, args: QueryKnowledgeBaseArgs) - str: 工具查询知识库。 # 这里可以直接使用注入的 kb_client无需在函数内部创建。 answer await kb_client.query(args.topic) return f关于【{args.topic}】知识库记录如下{answer} # 注册工具到 Agent下一节详述实操心得依赖注入的命名匹配是隐式的依赖于参数名。务必保持依赖函数返回的类型与工具函数或 Agent 的system_prompt中引用它的地方所期望的类型一致。一个良好的实践是为依赖返回值定义 Pydantic 模型或明确的类型别名如KnowledgeBaseClient这样类型检查器才能发挥作用提前发现错误。通过依赖注入我们将 Agent 的运行时环境谁在用、能访问什么资源与它的核心逻辑如何思考、如何使用工具清晰地分离开。这使得单元测试变得极其简单你可以轻松地为get_current_user依赖提供一个返回测试用户的函数而无需改动任何工具或 Agent 逻辑。5. 类型安全的工具注册与使用工具是 Agent 能力的延伸。PydanticAI 强制要求使用 Pydantic 模型来定义工具的参数这带来了无与伦比的类型安全和自文档化特性。我们来创建一个完整的工具并将其注册到 Agent。5.1 定义工具函数与参数模型我们接着上面的tool_query_knowledge_base函数来完善。通常我会把工具集中放在一个tools.py文件里。# tools.py from pydantic_ai import RunContext from pydantic import BaseModel, Field from typing import Dict, Any # 工具1查询知识库的参数模型 class QueryKBArgs(BaseModel): topic: str Field(description需要查询的知识库条目关键词例如退货政策、运费、客服时间) async def tool_query_kb(ctx: RunContext, kb_client, args: QueryKBArgs) - str: 查询内部知识库获取标准答案。 # 注意kb_client 参数名必须与某个依赖函数返回的客户端对象匹配。 # 这里假设 get_knowledge_base_conn 依赖返回了一个有 query 方法的对象。 answer await kb_client.query(args.topic) return answer # 工具2获取用户信息的参数模型虽然我们有依赖但有时也需要作为工具暴露给LLM class GetUserInfoArgs(BaseModel): # 这个工具可能不需要输入参数或者只需要一个字段来确认 confirm: bool Field(description确认为True时才返回用户信息, defaultTrue) async def tool_get_user_info(ctx: RunContext, current_user: Dict[str, Any], args: GetUserInfoArgs) - str: 获取当前用户的详细信息。 if not args.confirm: return 用户取消了信息获取请求。 user_info f用户ID: {current_user[id]}, 姓名: {current_user[name]}, VIP等级: {current_user[vip_level]} return user_info # 工具3一个计算器工具展示更复杂的参数类型 from typing import List class CalculatorArgs(BaseModel): operation: str Field(description运算类型只能是 add, subtract, multiply, divide 之一) numbers: List[float] Field(description参与运算的数字列表至少需要两个数字) async def tool_calculator(ctx: RunContext, args: CalculatorArgs) - float: 执行简单的数学运算。 if len(args.numbers) 2: raise ValueError(至少需要两个数字进行运算) result args.numbers[0] for num in args.numbers[1:]: if args.operation add: result num elif args.operation subtract: result - num elif args.operation multiply: result * num elif args.operation divide: if num 0: raise ValueError(除数不能为零) result / num else: raise ValueError(f不支持的运算类型: {args.operation}) return result5.2 将工具注册到 Agent注册工具让 LLM 知道它的存在、描述和调用格式。PydanticAI 提供了装饰器语法非常直观。# main.py 或 agent_builder.py from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from dependencies import get_current_user, get_knowledge_base_conn from tools import tool_query_kb, tool_get_user_info, tool_calculator from config import settings model OpenAIModel(...) # 创建 Agent 实例 agent Agent( modelmodel, system_prompt你是智能客服助手请根据知识库和用户信息回答问题。 你可以使用以下工具 1. tool_query_kb: 当你需要查询公司的标准政策、流程等信息时使用。 2. tool_get_user_info: 当用户询问关于其自身账户信息时使用。 3. tool_calculator: 当用户需要进行数学计算时使用。 请优先使用工具获取准确信息。, deps[get_knowledge_base_conn], # 默认依赖 ) # 使用装饰器注册工具 agent.tool async def query_kb(args: QueryKBArgs, ctx: RunContext, kb_client) - str: # 注意装饰器注册时函数签名可以更灵活但依赖参数kb_client必须存在。 return await tool_query_kb(ctx, kb_client, args) agent.tool async def get_user_info(args: GetUserInfoArgs, ctx: RunContext, current_user) - str: return await tool_get_user_info(ctx, current_user, args) agent.tool async def calculator(args: CalculatorArgs) - float: # 这个工具不需要依赖 return await tool_calculator(None, args) # 注意这里传入了 None 作为 ctx因为原函数需要它但实际未使用。 # 或者也可以使用 agent.tool 作为函数来注册如果你不想用装饰器 # agent.tool(tool_calculator, namecalc) # 可以指定工具名5.3 运行并观察工具调用现在让我们运行一个会触发工具调用的对话。async def main_with_tools(): user_query 我是VIP吗另外满50减10再打9折最后多少钱 result await agent.run( user_query, deps[get_current_user] # 注入用户依赖 ) print(f最终回复: {result.data}) print(\n 本次运行详情 ) print(f消耗Token: {result.usage}) print(f总成本: ${result.cost:.6f}) print(工具调用记录:) for call in result.tool_calls: print(f - 工具: {call.tool_name}, 参数: {call.args}, 结果: {call.result})运行这段代码你会看到 Agent 首先调用了get_user_info工具来确认用户 VIP 等级然后调用了calculator工具来计算(50 - 10) * 0.9的结果。所有的工具调用记录、参数和结果都清晰地保存在result.tool_calls里。注意事项工具函数的描述docstring和参数模型的字段描述Field(description...)非常重要LLM 主要依靠这些描述来决定是否以及如何调用工具。务必用清晰、准确的自然语言描述工具的功能和每个参数的含义。一个常见的坑是描述过于简略或模糊导致 LLM 不理解或错误调用。5.4 类型安全带来的好处你现在可以尝试在调用tool_calculator时传递错误的参数类型比如operation传一个数字或者numbers传一个字符串。如果你的 IDE 配置了类型检查如 Pylance, Pyright你会立刻看到波浪线错误提示。这就是类型安全在开发阶段带来的巨大优势将运行时错误提前到了编译或静态检查时。6. 实现流式输出从逐字生成到完整过程流流式输出是现代 AI 应用体验的关键。PydanticAI 的stream方法提供了强大的流式支持不仅能流式返回最终的文本还能流式返回工具调用的过程让我们可以构建出类似 ChatGPT 那样具有“思考过程”展示的交互界面。6.1 基础流式输出逐字接收最简单的流式就是接收 AI 生成的文本片段。async def basic_streaming(): user_query 用一段话介绍Python的Pydantic库。 # 关键使用 .stream() 而不是 .run() stream_result agent.stream(user_query) print(AI回复流式: , end, flushTrue) async for chunk in stream_result: # chunk 是一个 Result 对象但它的 data 属性是本次流片段新增的文本。 if chunk.data: print(chunk.data, end, flushTrue) # 逐字打印 print() # 换行在这个循环中chunk.data每次都是一小段新生成的文本。前端可以通过 Server-Sent Events (SSE) 或 WebSocket 将这些片段实时推送给用户。6.2 进阶捕获并流式输出工具调用过程仅仅流式文本还不够酷。我们更希望看到 Agent 的“思考链”它什么时候决定调用工具、调用了什么工具、工具返回了什么结果。PydanticAI 的流式结果中包含了丰富的元数据。async def advanced_streaming_with_tools(): user_query 我的VIP等级是多少然后计算一下(25 17) * 2 等于多少。 stream_result agent.stream( user_query, deps[get_current_user] ) async for chunk in stream_result: # 1. 输出新增的文本内容 if chunk.data: print(f[AI说]: {chunk.data}) # 2. 输出本次流中新增的工具调用Agent决定调用工具 # new_tool_calls 是本次 chunk 中新出现的工具调用列表 for tool_call in chunk.new_tool_calls: print(f[Agent决定调用工具] 工具名: {tool_call.tool_name}, 参数: {tool_call.args}) # 3. 输出本次流中完成的工具调用结果工具执行完毕 # tool_call_results 是本次 chunk 中完成的工具调用结果列表 for result in chunk.tool_call_results: print(f[工具执行结果] 工具: {result.tool_name}, 结果: {result.result}) # 4. 你还可以访问其他流式元数据如 chunk.usage (累计token消耗) # 注意流式下的 usage 和 cost 通常是累计值最终 chunk 的才是总值。运行这段代码你会看到类似下面的输出[Agent决定调用工具] 工具名: get_user_info, 参数: {confirm: True} [工具执行结果] 工具: get_user_info, 结果: 用户ID: user_001, 姓名: 张三, VIP等级: 3 [AI说]: 您的VIP等级是3级。 [Agent决定调用工具] 工具名: calculator, 参数: {operation: add, numbers: [25.0, 17.0]} [工具执行结果] 工具: calculator, 结果: 42.0 [Agent决定调用工具] 工具名: calculator, 参数: {operation: multiply, numbers: [42.0, 2.0]} [工具执行结果] 工具: calculator, 结果: 84.0 [AI说]: 接下来我为您计算 (25 17) * 2。首先25加17等于42。然后42乘以2等于84。所以最终结果是84。6.3 构建一个完整的流式响应处理器对于 Web 后端我们需要将这种流式结构转化为前端友好的格式通常是 JSON 序列化的事件流。下面是一个模拟的处理器函数它展示了如何分类处理不同的流事件import json from enum import Enum from pydantic_ai import Agent class StreamEventType(str, Enum): TEXT text TOOL_CALL tool_call TOOL_RESULT tool_result ERROR error DONE done async def generate_agent_stream(agent: Agent, query: str, depsNone): 生成一个标准化的流式事件序列适用于SSE。 这是一个生成器函数yield 出 JSON 字符串。 try: stream_result agent.stream(query, depsdeps or []) async for chunk in stream_result: # 事件1: 文本内容 if chunk.data: yield json.dumps({ type: StreamEventType.TEXT, content: chunk.data }) \n\n # 事件2: 新的工具调用 for tool_call in chunk.new_tool_calls: yield json.dumps({ type: StreamEventType.TOOL_CALL, tool_name: tool_call.tool_name, args: tool_call.args }) \n\n # 事件3: 工具调用结果 for tool_result in chunk.tool_call_results: yield json.dumps({ type: StreamEventType.TOOL_RESULT, tool_name: tool_result.tool_name, result: tool_result.result }) \n\n # 事件4: 流结束 yield json.dumps({ type: StreamEventType.DONE, final_usage: chunk.usage, # 最后一个chunk包含总的usage final_cost: chunk.cost }) \n\n except Exception as e: # 事件5: 错误 yield json.dumps({ type: StreamEventType.ERROR, error: str(e) }) \n\n # 在异步Web框架如FastAPI中的使用示例 from fastapi import FastAPI, Response from fastapi.responses import StreamingResponse app FastAPI() app.get(/chat/stream) async def chat_stream(query: str): async def event_generator(): async for event in generate_agent_stream(agent, query, deps[get_current_user]): yield event return StreamingResponse( event_generator(), media_typetext/event-stream, headers{Cache-Control: no-cache, Connection: keep-alive} )实操心得处理流式输出时一定要注意错误处理。网络中断、工具函数抛出异常、LLM API 限流等都可能导致流中断。务必在async for循环外包裹try...except并将错误信息作为一个特定的事件类型如ERRORyield 出去让前端能优雅地处理并提示用户。此外流式响应要设置正确的 HTTP 头如text/event-stream并确保连接保持活跃。7. 常见问题、调试技巧与性能优化在实际开发和部署中你肯定会遇到各种问题。这里我总结了一些高频问题和处理技巧。7.1 工具不被调用或调用错误问题LLM 似乎“忽略”了某个工具或者在应该调用时没有调用。排查检查工具描述首先检查工具函数的 docstring 和参数模型的字段描述是否清晰、无歧义。LLM 完全依赖这些描述做决策。描述要具体比如“查询用户订单状态”就比“获取订单信息”好。检查系统提示词在Agent的system_prompt中你是否明确告知了 Agent 可以使用哪些工具以及何时使用一个好的实践是在提示词中列出工具名和简短用途。查看原始消息PydanticAI 的Result对象有result.messages属性里面包含了完整的对话历史包括 AI 决定调用工具的“函数调用”消息。打印出来看看 LLM 是否生成了正确的工具调用请求。启用调试日志PydanticAI 和底层的 OpenAI SDK 通常有日志功能。设置logging级别为DEBUG可以查看详细的请求和响应 JSON这对于理解 LLM 的“思考”过程非常有帮助。7.2 依赖注入失败或类型错误问题工具函数报错提示某个依赖参数找不到TypeError或者运行时类型不匹配。排查参数名匹配确保工具函数中接收依赖的参数名与依赖函数返回值的“标识”完全一致。默认情况下标识就是依赖函数的函数名。例如依赖函数async def get_db()工具函数参数就应该是db。显式指定依赖你可以在agent.tool装饰器或agent.tool()注册函数时通过deps参数显式指定该工具需要哪些依赖覆盖默认的命名匹配。例如agent.tool(deps[get_db])。使用类型注解为工具函数的依赖参数加上明确的类型注解如db: DatabaseClient。这虽然不影响运行时但能让 mypy 等工具在静态检查时发现问题。7.3 流式输出中断或不完整问题流式响应突然停止前端收不到DONE事件或者工具调用结果丢失。排查超时设置检查你的 HTTP 服务器、反向代理如 Nginx以及 LLM API 客户端的超时设置。流式响应可能持续很长时间需要将超时时间设置得足够长或者禁用超时。异常捕获如 6.3 节所述务必在流式生成器的外层进行全面的异常捕获并将错误信息 yield 出去。一个未捕获的异常会导致连接突然关闭。检查工具函数确保你的工具函数是异步的async def且自身是健壮的。一个同步的、耗时的或会抛出异常的工具函数会阻塞整个流。网络稳定性对于生产环境考虑加入重试逻辑特别是在调用外部 API 的工具中和使用连接池。7.4 性能优化建议依赖缓存如果某个依赖的获取成本很高比如查询数据库且在同一轮对话中多次运行 Agent 时不会改变可以考虑使用agent.dep装饰器并设置cacheTrue。PydanticAI 会在单次run或stream调用期间缓存该依赖的结果。from pydantic_ai import Agent agent Agent(...) agent.dep(cacheTrue) # 这个依赖的结果会被缓存 async def get_expensive_config(): # 模拟一个耗时的配置读取 await asyncio.sleep(1) return {key: value}合理设置max_steps在创建Agent时可以设置max_steps参数它限制了一轮对话中 Agent 进行“思考-行动工具调用”循环的最大次数。防止 Agent 陷入无限循环或执行过多不必要的工具调用。默认值通常是 10对于复杂任务可能需要调高。批量处理工具调用如果 Agent 有可能在一次推理中提出多个并行的工具调用请求LLM 支持 function calling 的并行调用确保你的工具函数和下游服务能够处理并发。合理使用asyncio.gather来并行执行多个独立的工具调用可以显著减少整体响应时间。监控与限流记录每次运行的result.usage和result.cost以便监控成本和用量。对于公开服务实施基于用户或 IP 的速率限制防止滥用。通过系统地应用这些调试方法和优化策略你的类型安全 AI Agent 将不仅健壮可靠还能在高并发场景下保持良好的性能表现。这其中的很多经验都是在真实项目踩坑后总结出来的希望你能避开这些陷阱。 基于PydanticAI构建类型安全的AI Agent：依赖注入、工具注册与流式输出实战
1. 项目概述为什么我们需要一个类型安全的 AI Agent最近在捣鼓 AI Agent 开发发现一个挺普遍的问题代码写着写着就成了一团乱麻。特别是当你需要集成多个工具、管理复杂的对话状态还要处理流式输出时那种感觉就像在指挥一支没有经过任何训练的乐队每个乐手模块都在即兴发挥最后出来的声音可想而知。传统的开发方式比如直接用 OpenAI SDK 或者一些早期框架往往把 API 调用、工具调用逻辑、状态管理全部揉在一起缺乏清晰的边界和类型约束。这就导致两个大问题一是调试困难一个参数类型传错了可能要等到运行时才能发现二是代码难以维护和扩展加个新功能就得小心翼翼生怕碰坏了其他地方。这时候PydanticAI进入了我的视野。它不是一个全新的概念而是建立在 Pydantic 这个强大的数据验证库和 Python 的类型提示Type Hints体系之上专门为构建 AI Agent 而生。它的核心卖点就是类型安全Type Safety。简单来说它让你能用写 Python 类和方法的那种清晰、可预测的方式来定义 Agent 的行为、工具的参数和返回结果。编译器或类型检查器如 mypy能在你写代码的时候就帮你揪出很多潜在的错误而不是等到程序跑起来再报一堆让人头疼的异常。这个项目就是带你从零开始用 PydanticAI 搭建一个具备完整功能的 AI Agent。我们会重点攻克几个在实际开发中高频出现的关键环节如何优雅地使用依赖注入来管理 Agent 所需的上下文比如数据库连接、配置信息而不是硬编码在函数里如何规范地注册和使用工具确保工具的参数和返回值都有明确的类型定义最后如何实现流式输出让 Agent 的思考过程或最终答案能够像 ChatGPT 那样一个字一个字地“流”出来提升用户体验。整个过程你会看到类型安全如何让 Agent 的开发从“刀耕火种”走向“精耕细作”。2. 核心设计思路依赖注入、工具与流的三角架构在动手写代码之前我们先得把架构想清楚。一个健壮的 AI Agent尤其是业务逻辑稍复杂的不能把所有东西都塞进一个巨大的main函数里。PydanticAI 鼓励我们采用一种更模块化、更易于测试的设计。我把它总结为“三角架构”三个顶点分别是依赖Dependencies、工具Tools和执行流Execution Streaming。2.1 依赖注入给 Agent 提供“上下文装备”依赖注入听起来高大上其实理念很简单不是让 Agent 自己内部去创建它需要的东西比如数据库客户端、配置对象、其他服务而是由外部“注入”给它。这样做的好处太多了解耦Agent 的逻辑不依赖于具体的实现。今天用 SQLite明天换 PostgreSQL只需要换一个注入的客户端Agent 代码一行不用改。可测试性测试时你可以轻松地注入一个“模拟Mock”的数据库客户端而不是去连接一个真实的数据库。状态管理像用户会话、长期记忆这些状态可以通过依赖来管理和共享。在 PydanticAI 中依赖通常被定义为异步函数async def它们的返回值会被自动提供给 Agent 的run方法或工具函数。例如一个获取当前用户信息的依赖一个获取数据库连接的依赖。2.2 工具注册定义 Agent 的“手脚”工具是 Agent 与外部世界交互的桥梁。查天气、发邮件、读写数据库都需要通过工具。PydanticAI 要求我们用 Pydantic 模型来严格定义工具的输入参数args_schema这直接带来了类型安全。你在代码里调用工具时如果参数类型不对IDE 会直接报错。工具注册的本质是将这些定义好的工具函数“告诉” Agent。PydanticAI 提供了装饰器如agent.tool和显式注册等多种方式。注册后的工具其描述和参数 schema 会被自动编入发送给大语言模型LLM的提示词Prompt中LLM 才知道在什么情况下、以什么格式来调用这个工具。2.3 流式输出让交互过程“活”起来流式输出是现代 AI 应用的标配。它不仅仅是把最终答案一次性返回更重要的是可以实时返回 Agent 的“思考过程”Reasoning或部分结果。这对于需要长时间运行的任务如编写代码、分析长文档至关重要用户不用傻等着能看到进度。PydanticAI 原生支持流式输出。其核心在于stream方法它返回的是一个异步生成器AsyncGenerator。这个生成器不仅会产出最终的答案还会产出中间的关键步骤比如“调用了哪个工具”、“工具返回了什么结果”、“Agent 现在在想什么”。我们需要做的就是处理好这个流将其转化为前端可以消费的格式如 Server-Sent Events。这个三角架构是环环相扣的依赖为工具和 Agent 提供运行环境工具扩展了 Agent 的能力边界流式输出则将依赖和工具协作产生的过程与结果高效地呈现给用户。接下来我们就进入实战环节一步步把它们搭建起来。3. 环境准备与 PydanticAI 核心概念速览工欲善其事必先利其器。我们先来把开发环境搭好并快速理解几个 PydanticAI 的核心模型Model这是后续所有工作的基础。3.1 创建项目与安装依赖我习惯用一个干净的虚拟环境来开始新项目避免包版本冲突。# 创建项目目录并进入 mkdir type-safe-ai-agent cd type-safe-ai-agent # 创建虚拟环境这里用 venv你也可以用 conda python -m venv .venv # 激活虚拟环境 # Windows: .venv\Scripts\activate # Linux/Mac: source .venv/bin/activate接下来安装核心依赖。除了pydantic-ai我们还需要openai作为 LLM 后端当然 PydanticAI 也支持 Anthropic、Gemini 等以及httpx用于可能的网络请求pydantic-settings用来管理配置。pip install pydantic-ai openai httpx pydantic-settings # 可选用于环境变量管理 pip install python-dotenv3.2 理解 PydanticAI 的四大核心模型PydanticAI 的 API 设计围绕着几个核心类展开理解它们的关系至关重要。Agent代理这是最外层的容器代表一个 AI 代理。它绑定了一个具体的 LLM如 GPT-4并包含了一系列的工具、系统提示词System Prompt和运行配置。我们通过实例化Agent类来创建代理。Model模型在 PydanticAI 上下文中Model类是对 LLM 客户端如 OpenAI client的封装。它处理与 LLM API 的实际通信包括格式化请求、解析响应、处理错误等。创建Agent时必须指定一个Model。RunContext运行上下文这是 Agent 单次运行run或stream的上下文环境。它包含了本次运行的所有状态信息比如用户输入messages、注入的依赖结果deps、可用的工具列表等。我们通常不需要直接创建它但会在工具函数和依赖函数中通过参数访问它。Result结果Agent 运行后返回的对象。它包含了最核心的 AI 回复内容data以及完整的运行元数据比如调用了哪些工具tool_calls、消耗的 Token 数usage、本次运行的成本cost等。对于流式输出我们收到的是多个Result对象每个对象代表流中的一个“块”chunk。一个最简单的非流式 Agent 调用流程是这样的你创建Model和Agent然后调用agent.run(“用户问题”)返回一个Result对象从result.data里拿到 AI 的回复。3.3 初始化配置与 Model在实际项目中API Key 等敏感信息肯定不能写死在代码里。我们用pydantic-settings和.env文件来管理。首先在项目根目录创建.env文件OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果你用官方API # 如果使用其他兼容OpenAI的API服务可以修改 BASE_URL # OPENAI_BASE_URLhttps://your.proxy.com/v1然后创建一个config.py文件来定义配置from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str openai_base_url: str https://api.openai.com/v1 # 可以添加其他配置如模型名称 model_name: str gpt-4o-mini class Config: env_file .env settings Settings()接着在main.py或你的应用入口文件中初始化 Modelimport asyncio from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from config import settings # 1. 创建 Model这是与LLM通信的客户端 model OpenAIModel( model_namesettings.model_name, api_keysettings.openai_api_key, base_urlsettings.openai_base_url, ) # 2. 创建 Agent绑定 Model 并设置系统提示词 agent Agent( modelmodel, system_prompt你是一个乐于助人的AI助手。请用中文回答用户的问题。, ) async def main(): # 3. 运行 Agent result await agent.run(你好世界) print(fAI回复: {result.data}) if __name__ __main__: asyncio.run(main())运行这个脚本如果一切配置正确你应该能看到 AI 的问候回复。这标志着你的 PydanticAI 基础环境已经跑通了。接下来我们要为这个 Agent 注入灵魂——依赖和工具。4. 实战依赖注入构建可测试的 Agent 上下文现在我们来给 Agent 加上“依赖”。假设我们要构建一个客服 Agent它需要知道当前咨询的用户是谁用户信息并且能够访问知识库数据库连接。我们不会把这些信息硬编码在工具函数里而是通过依赖注入。4.1 定义依赖函数依赖函数就是普通的async def函数它的参数可以接收RunContext。PydanticAI 会自动调用这些函数并将它们的返回值放入上下文中供后续的 Agent 逻辑或工具使用。我们在项目里创建一个dependencies.py文件from pydantic_ai import RunContext from typing import Dict, Any import httpx # 假设的用户数据库实际项目中可能是Redis、SQL数据库 fake_user_db { user_001: {id: user_001, name: 张三, vip_level: 3}, user_002: {id: user_002, name: 李四, vip_level: 1}, } async def get_current_user(ctx: RunContext) - Dict[str, Any]: 依赖1获取当前用户信息。 在实际Web应用中ctx 里可能包含了从请求头中解析出的用户令牌token。 这里我们模拟从‘ctx’的某个属性比如 ctx.state中获取用户ID。 # 模拟假设我们从上下文的某个地方拿到了用户ID。这里为了演示我们写死一个。 # 真实场景下可能是 user_id ctx.state.get(user_id) user_id user_001 user fake_user_db.get(user_id) if not user: # 如果用户不存在可以返回一个默认用户或抛出错误 # PydanticAI 支持依赖抛出异常这会导致整个 Agent 运行失败 raise ValueError(f用户 {user_id} 不存在) return user async def get_knowledge_base_conn(ctx: RunContext): 依赖2获取知识库连接。 这里我们模拟一个简单的键值存储客户端。实际可能是 Elasticsearch、向量数据库等的连接。 注意这个函数返回的不是数据而是一个“客户端”对象。 # 模拟一个简单的内存知识库 class KnowledgeBaseClient: def __init__(self): self.data { 退货政策: 商品签收后7天内可无理由退货保持商品完好。, 运费说明: 订单满99元包邮不满99元收取10元运费。, 客服时间: 人工客服工作时间周一至周日 9:00-18:00。, } async def query(self, topic: str) - str: return self.data.get(topic, 抱歉未找到相关知识点。) # 每次调用依赖返回一个新的客户端实例或共享的连接池 # 对于数据库连接更常见的做法是注入一个连接池这里简化处理。 return KnowledgeBaseClient() # 可以定义更多依赖如获取系统配置、日志记录器等。4.2 在 Agent 中使用依赖定义了依赖函数后我们需要在创建 Agent 或运行 Agent 时声明它们。有两种主要方式方式一在Agent.run()或Agent.stream()时通过deps参数注入。这种方式灵活适合依赖项在每次运行时可能不同的场景。# 接之前的 main 函数 async def main_with_deps(): user_question 你们的退货政策是怎样的 result await agent.run( user_question, deps[get_current_user, get_knowledge_base_conn] # 传入依赖函数列表 ) print(fAI回复: {result.data}) # 我们还可以从 result 里查看本次运行使用的依赖值调试用 print(f本次运行注入的用户是: {result.deps_results[get_current_user]})方式二将依赖定义为 Agent 的默认依赖。这种方式适合那些全局的、每次运行都需要的依赖比如配置、日志器。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from dependencies import get_knowledge_base_conn from config import settings # 创建 Model (同上) model OpenAIModel(...) # 创建 Agent 时指定默认依赖 agent_with_default_deps Agent( modelmodel, system_prompt你是智能客服助手请根据知识库和用户信息回答问题。, deps[get_knowledge_base_conn], # 默认依赖每次运行都会自动注入 ) async def main_default_deps(): # 这次运行只需要再注入用户依赖即可 result await agent_with_default_deps.run( 我想查询运费。, deps[get_current_user] # 合并默认依赖和本次运行依赖 ) print(result.data)4.3 在工具函数中访问依赖这是依赖注入价值体现的关键点。工具函数可以通过参数直接访问由依赖注入的值。PydanticAI 会自动根据参数名进行匹配。from pydantic_ai import RunContext from pydantic import BaseModel, Field # 首先定义一个工具的输入参数模型 class QueryKnowledgeBaseArgs(BaseModel): topic: str Field(description要查询的知识点主题例如‘退货政策’、‘运费说明’) # 注意工具函数的参数ctx 是 RunContextkb_client 就是依赖 get_knowledge_base_conn 的返回值。 # PydanticAI 会自动将同名的依赖结果传递进来。 async def tool_query_knowledge_base(ctx: RunContext, kb_client, args: QueryKnowledgeBaseArgs) - str: 工具查询知识库。 # 这里可以直接使用注入的 kb_client无需在函数内部创建。 answer await kb_client.query(args.topic) return f关于【{args.topic}】知识库记录如下{answer} # 注册工具到 Agent下一节详述实操心得依赖注入的命名匹配是隐式的依赖于参数名。务必保持依赖函数返回的类型与工具函数或 Agent 的system_prompt中引用它的地方所期望的类型一致。一个良好的实践是为依赖返回值定义 Pydantic 模型或明确的类型别名如KnowledgeBaseClient这样类型检查器才能发挥作用提前发现错误。通过依赖注入我们将 Agent 的运行时环境谁在用、能访问什么资源与它的核心逻辑如何思考、如何使用工具清晰地分离开。这使得单元测试变得极其简单你可以轻松地为get_current_user依赖提供一个返回测试用户的函数而无需改动任何工具或 Agent 逻辑。5. 类型安全的工具注册与使用工具是 Agent 能力的延伸。PydanticAI 强制要求使用 Pydantic 模型来定义工具的参数这带来了无与伦比的类型安全和自文档化特性。我们来创建一个完整的工具并将其注册到 Agent。5.1 定义工具函数与参数模型我们接着上面的tool_query_knowledge_base函数来完善。通常我会把工具集中放在一个tools.py文件里。# tools.py from pydantic_ai import RunContext from pydantic import BaseModel, Field from typing import Dict, Any # 工具1查询知识库的参数模型 class QueryKBArgs(BaseModel): topic: str Field(description需要查询的知识库条目关键词例如退货政策、运费、客服时间) async def tool_query_kb(ctx: RunContext, kb_client, args: QueryKBArgs) - str: 查询内部知识库获取标准答案。 # 注意kb_client 参数名必须与某个依赖函数返回的客户端对象匹配。 # 这里假设 get_knowledge_base_conn 依赖返回了一个有 query 方法的对象。 answer await kb_client.query(args.topic) return answer # 工具2获取用户信息的参数模型虽然我们有依赖但有时也需要作为工具暴露给LLM class GetUserInfoArgs(BaseModel): # 这个工具可能不需要输入参数或者只需要一个字段来确认 confirm: bool Field(description确认为True时才返回用户信息, defaultTrue) async def tool_get_user_info(ctx: RunContext, current_user: Dict[str, Any], args: GetUserInfoArgs) - str: 获取当前用户的详细信息。 if not args.confirm: return 用户取消了信息获取请求。 user_info f用户ID: {current_user[id]}, 姓名: {current_user[name]}, VIP等级: {current_user[vip_level]} return user_info # 工具3一个计算器工具展示更复杂的参数类型 from typing import List class CalculatorArgs(BaseModel): operation: str Field(description运算类型只能是 add, subtract, multiply, divide 之一) numbers: List[float] Field(description参与运算的数字列表至少需要两个数字) async def tool_calculator(ctx: RunContext, args: CalculatorArgs) - float: 执行简单的数学运算。 if len(args.numbers) 2: raise ValueError(至少需要两个数字进行运算) result args.numbers[0] for num in args.numbers[1:]: if args.operation add: result num elif args.operation subtract: result - num elif args.operation multiply: result * num elif args.operation divide: if num 0: raise ValueError(除数不能为零) result / num else: raise ValueError(f不支持的运算类型: {args.operation}) return result5.2 将工具注册到 Agent注册工具让 LLM 知道它的存在、描述和调用格式。PydanticAI 提供了装饰器语法非常直观。# main.py 或 agent_builder.py from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from dependencies import get_current_user, get_knowledge_base_conn from tools import tool_query_kb, tool_get_user_info, tool_calculator from config import settings model OpenAIModel(...) # 创建 Agent 实例 agent Agent( modelmodel, system_prompt你是智能客服助手请根据知识库和用户信息回答问题。 你可以使用以下工具 1. tool_query_kb: 当你需要查询公司的标准政策、流程等信息时使用。 2. tool_get_user_info: 当用户询问关于其自身账户信息时使用。 3. tool_calculator: 当用户需要进行数学计算时使用。 请优先使用工具获取准确信息。, deps[get_knowledge_base_conn], # 默认依赖 ) # 使用装饰器注册工具 agent.tool async def query_kb(args: QueryKBArgs, ctx: RunContext, kb_client) - str: # 注意装饰器注册时函数签名可以更灵活但依赖参数kb_client必须存在。 return await tool_query_kb(ctx, kb_client, args) agent.tool async def get_user_info(args: GetUserInfoArgs, ctx: RunContext, current_user) - str: return await tool_get_user_info(ctx, current_user, args) agent.tool async def calculator(args: CalculatorArgs) - float: # 这个工具不需要依赖 return await tool_calculator(None, args) # 注意这里传入了 None 作为 ctx因为原函数需要它但实际未使用。 # 或者也可以使用 agent.tool 作为函数来注册如果你不想用装饰器 # agent.tool(tool_calculator, namecalc) # 可以指定工具名5.3 运行并观察工具调用现在让我们运行一个会触发工具调用的对话。async def main_with_tools(): user_query 我是VIP吗另外满50减10再打9折最后多少钱 result await agent.run( user_query, deps[get_current_user] # 注入用户依赖 ) print(f最终回复: {result.data}) print(\n 本次运行详情 ) print(f消耗Token: {result.usage}) print(f总成本: ${result.cost:.6f}) print(工具调用记录:) for call in result.tool_calls: print(f - 工具: {call.tool_name}, 参数: {call.args}, 结果: {call.result})运行这段代码你会看到 Agent 首先调用了get_user_info工具来确认用户 VIP 等级然后调用了calculator工具来计算(50 - 10) * 0.9的结果。所有的工具调用记录、参数和结果都清晰地保存在result.tool_calls里。注意事项工具函数的描述docstring和参数模型的字段描述Field(description...)非常重要LLM 主要依靠这些描述来决定是否以及如何调用工具。务必用清晰、准确的自然语言描述工具的功能和每个参数的含义。一个常见的坑是描述过于简略或模糊导致 LLM 不理解或错误调用。5.4 类型安全带来的好处你现在可以尝试在调用tool_calculator时传递错误的参数类型比如operation传一个数字或者numbers传一个字符串。如果你的 IDE 配置了类型检查如 Pylance, Pyright你会立刻看到波浪线错误提示。这就是类型安全在开发阶段带来的巨大优势将运行时错误提前到了编译或静态检查时。6. 实现流式输出从逐字生成到完整过程流流式输出是现代 AI 应用体验的关键。PydanticAI 的stream方法提供了强大的流式支持不仅能流式返回最终的文本还能流式返回工具调用的过程让我们可以构建出类似 ChatGPT 那样具有“思考过程”展示的交互界面。6.1 基础流式输出逐字接收最简单的流式就是接收 AI 生成的文本片段。async def basic_streaming(): user_query 用一段话介绍Python的Pydantic库。 # 关键使用 .stream() 而不是 .run() stream_result agent.stream(user_query) print(AI回复流式: , end, flushTrue) async for chunk in stream_result: # chunk 是一个 Result 对象但它的 data 属性是本次流片段新增的文本。 if chunk.data: print(chunk.data, end, flushTrue) # 逐字打印 print() # 换行在这个循环中chunk.data每次都是一小段新生成的文本。前端可以通过 Server-Sent Events (SSE) 或 WebSocket 将这些片段实时推送给用户。6.2 进阶捕获并流式输出工具调用过程仅仅流式文本还不够酷。我们更希望看到 Agent 的“思考链”它什么时候决定调用工具、调用了什么工具、工具返回了什么结果。PydanticAI 的流式结果中包含了丰富的元数据。async def advanced_streaming_with_tools(): user_query 我的VIP等级是多少然后计算一下(25 17) * 2 等于多少。 stream_result agent.stream( user_query, deps[get_current_user] ) async for chunk in stream_result: # 1. 输出新增的文本内容 if chunk.data: print(f[AI说]: {chunk.data}) # 2. 输出本次流中新增的工具调用Agent决定调用工具 # new_tool_calls 是本次 chunk 中新出现的工具调用列表 for tool_call in chunk.new_tool_calls: print(f[Agent决定调用工具] 工具名: {tool_call.tool_name}, 参数: {tool_call.args}) # 3. 输出本次流中完成的工具调用结果工具执行完毕 # tool_call_results 是本次 chunk 中完成的工具调用结果列表 for result in chunk.tool_call_results: print(f[工具执行结果] 工具: {result.tool_name}, 结果: {result.result}) # 4. 你还可以访问其他流式元数据如 chunk.usage (累计token消耗) # 注意流式下的 usage 和 cost 通常是累计值最终 chunk 的才是总值。运行这段代码你会看到类似下面的输出[Agent决定调用工具] 工具名: get_user_info, 参数: {confirm: True} [工具执行结果] 工具: get_user_info, 结果: 用户ID: user_001, 姓名: 张三, VIP等级: 3 [AI说]: 您的VIP等级是3级。 [Agent决定调用工具] 工具名: calculator, 参数: {operation: add, numbers: [25.0, 17.0]} [工具执行结果] 工具: calculator, 结果: 42.0 [Agent决定调用工具] 工具名: calculator, 参数: {operation: multiply, numbers: [42.0, 2.0]} [工具执行结果] 工具: calculator, 结果: 84.0 [AI说]: 接下来我为您计算 (25 17) * 2。首先25加17等于42。然后42乘以2等于84。所以最终结果是84。6.3 构建一个完整的流式响应处理器对于 Web 后端我们需要将这种流式结构转化为前端友好的格式通常是 JSON 序列化的事件流。下面是一个模拟的处理器函数它展示了如何分类处理不同的流事件import json from enum import Enum from pydantic_ai import Agent class StreamEventType(str, Enum): TEXT text TOOL_CALL tool_call TOOL_RESULT tool_result ERROR error DONE done async def generate_agent_stream(agent: Agent, query: str, depsNone): 生成一个标准化的流式事件序列适用于SSE。 这是一个生成器函数yield 出 JSON 字符串。 try: stream_result agent.stream(query, depsdeps or []) async for chunk in stream_result: # 事件1: 文本内容 if chunk.data: yield json.dumps({ type: StreamEventType.TEXT, content: chunk.data }) \n\n # 事件2: 新的工具调用 for tool_call in chunk.new_tool_calls: yield json.dumps({ type: StreamEventType.TOOL_CALL, tool_name: tool_call.tool_name, args: tool_call.args }) \n\n # 事件3: 工具调用结果 for tool_result in chunk.tool_call_results: yield json.dumps({ type: StreamEventType.TOOL_RESULT, tool_name: tool_result.tool_name, result: tool_result.result }) \n\n # 事件4: 流结束 yield json.dumps({ type: StreamEventType.DONE, final_usage: chunk.usage, # 最后一个chunk包含总的usage final_cost: chunk.cost }) \n\n except Exception as e: # 事件5: 错误 yield json.dumps({ type: StreamEventType.ERROR, error: str(e) }) \n\n # 在异步Web框架如FastAPI中的使用示例 from fastapi import FastAPI, Response from fastapi.responses import StreamingResponse app FastAPI() app.get(/chat/stream) async def chat_stream(query: str): async def event_generator(): async for event in generate_agent_stream(agent, query, deps[get_current_user]): yield event return StreamingResponse( event_generator(), media_typetext/event-stream, headers{Cache-Control: no-cache, Connection: keep-alive} )实操心得处理流式输出时一定要注意错误处理。网络中断、工具函数抛出异常、LLM API 限流等都可能导致流中断。务必在async for循环外包裹try...except并将错误信息作为一个特定的事件类型如ERRORyield 出去让前端能优雅地处理并提示用户。此外流式响应要设置正确的 HTTP 头如text/event-stream并确保连接保持活跃。7. 常见问题、调试技巧与性能优化在实际开发和部署中你肯定会遇到各种问题。这里我总结了一些高频问题和处理技巧。7.1 工具不被调用或调用错误问题LLM 似乎“忽略”了某个工具或者在应该调用时没有调用。排查检查工具描述首先检查工具函数的 docstring 和参数模型的字段描述是否清晰、无歧义。LLM 完全依赖这些描述做决策。描述要具体比如“查询用户订单状态”就比“获取订单信息”好。检查系统提示词在Agent的system_prompt中你是否明确告知了 Agent 可以使用哪些工具以及何时使用一个好的实践是在提示词中列出工具名和简短用途。查看原始消息PydanticAI 的Result对象有result.messages属性里面包含了完整的对话历史包括 AI 决定调用工具的“函数调用”消息。打印出来看看 LLM 是否生成了正确的工具调用请求。启用调试日志PydanticAI 和底层的 OpenAI SDK 通常有日志功能。设置logging级别为DEBUG可以查看详细的请求和响应 JSON这对于理解 LLM 的“思考”过程非常有帮助。7.2 依赖注入失败或类型错误问题工具函数报错提示某个依赖参数找不到TypeError或者运行时类型不匹配。排查参数名匹配确保工具函数中接收依赖的参数名与依赖函数返回值的“标识”完全一致。默认情况下标识就是依赖函数的函数名。例如依赖函数async def get_db()工具函数参数就应该是db。显式指定依赖你可以在agent.tool装饰器或agent.tool()注册函数时通过deps参数显式指定该工具需要哪些依赖覆盖默认的命名匹配。例如agent.tool(deps[get_db])。使用类型注解为工具函数的依赖参数加上明确的类型注解如db: DatabaseClient。这虽然不影响运行时但能让 mypy 等工具在静态检查时发现问题。7.3 流式输出中断或不完整问题流式响应突然停止前端收不到DONE事件或者工具调用结果丢失。排查超时设置检查你的 HTTP 服务器、反向代理如 Nginx以及 LLM API 客户端的超时设置。流式响应可能持续很长时间需要将超时时间设置得足够长或者禁用超时。异常捕获如 6.3 节所述务必在流式生成器的外层进行全面的异常捕获并将错误信息 yield 出去。一个未捕获的异常会导致连接突然关闭。检查工具函数确保你的工具函数是异步的async def且自身是健壮的。一个同步的、耗时的或会抛出异常的工具函数会阻塞整个流。网络稳定性对于生产环境考虑加入重试逻辑特别是在调用外部 API 的工具中和使用连接池。7.4 性能优化建议依赖缓存如果某个依赖的获取成本很高比如查询数据库且在同一轮对话中多次运行 Agent 时不会改变可以考虑使用agent.dep装饰器并设置cacheTrue。PydanticAI 会在单次run或stream调用期间缓存该依赖的结果。from pydantic_ai import Agent agent Agent(...) agent.dep(cacheTrue) # 这个依赖的结果会被缓存 async def get_expensive_config(): # 模拟一个耗时的配置读取 await asyncio.sleep(1) return {key: value}合理设置max_steps在创建Agent时可以设置max_steps参数它限制了一轮对话中 Agent 进行“思考-行动工具调用”循环的最大次数。防止 Agent 陷入无限循环或执行过多不必要的工具调用。默认值通常是 10对于复杂任务可能需要调高。批量处理工具调用如果 Agent 有可能在一次推理中提出多个并行的工具调用请求LLM 支持 function calling 的并行调用确保你的工具函数和下游服务能够处理并发。合理使用asyncio.gather来并行执行多个独立的工具调用可以显著减少整体响应时间。监控与限流记录每次运行的result.usage和result.cost以便监控成本和用量。对于公开服务实施基于用户或 IP 的速率限制防止滥用。通过系统地应用这些调试方法和优化策略你的类型安全 AI Agent 将不仅健壮可靠还能在高并发场景下保持良好的性能表现。这其中的很多经验都是在真实项目踩坑后总结出来的希望你能避开这些陷阱。