当前位置：首页 > news >正文

基于PydanticAI构建类型安全的AI Agent：依赖注入、工具注册与流式输出实战

news 2026/8/14 7:03:56

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-dotenv

3.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_KEY=sk-your-openai-api-key-here OPENAI_BASE_URL=https://api.openai.com/v1 # 如果你用官方API # 如果使用其他兼容OpenAI的API服务，可以修改 BASE_URL # OPENAI_BASE_URL=https://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或你的应用入口文件中初始化 Model：

import asyncio from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from config import settings # 1. 创建 Model，这是与LLM通信的客户端 model = OpenAIModel( model_name=settings.model_name, api_key=settings.openai_api_key, base_url=settings.openai_base_url, ) # 2. 创建 Agent，绑定 Model 并设置系统提示词 agent = Agent( model=model, system_prompt="你是一个乐于助人的AI助手。请用中文回答用户的问题。", ) async def main(): # 3. 运行 Agent result = await agent.run("你好，世界！") print(f"AI回复: {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(f"AI回复: {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( model=model, 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` 是 RunContext，`kb_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时，才返回用户信息", default=True) 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 result

5.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( model=model, 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, name="calc") # 可以指定工具名

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="", flush=True) async for chunk in stream_result: # chunk 是一个 Result 对象，但它的 `data` 属性是本次流片段新增的文本。 if chunk.data: print(chunk.data, end="", flush=True) # 逐字打印 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, deps=None): """ 生成一个标准化的流式事件序列，适用于SSE。 这是一个生成器函数，yield 出 JSON 字符串。 """ try: stream_result = agent.stream(query, deps=deps 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_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"} )

实操心得：处理流式输出时，一定要注意错误处理。网络中断、工具函数抛出异常、LLM API 限流等都可能导致流中断。务必在async for循环外包裹try...except，并将错误信息作为一个特定的事件类型（如ERROR）yield 出去，让前端能优雅地处理并提示用户。此外，流式响应要设置正确的 HTTP 头（如text/event-stream），并确保连接保持活跃。

7. 常见问题、调试技巧与性能优化

在实际开发和部署中，你肯定会遇到各种问题。这里我总结了一些高频问题和处理技巧。

7.1 工具不被调用或调用错误

问题：LLM 似乎“忽略”了某个工具，或者在应该调用时没有调用。
排查：
1. 检查工具描述：首先，检查工具函数的 docstring 和参数模型的字段描述是否清晰、无歧义。LLM 完全依赖这些描述做决策。描述要具体，比如“查询用户订单状态”就比“获取订单信息”好。
2. 检查系统提示词：在Agent的system_prompt中，你是否明确告知了 Agent 可以使用哪些工具以及何时使用？一个好的实践是在提示词中列出工具名和简短用途。
3. 查看原始消息：PydanticAI 的Result对象有result.messages属性，里面包含了完整的对话历史，包括 AI 决定调用工具的“函数调用”消息。打印出来看看 LLM 是否生成了正确的工具调用请求。
4. 启用调试日志：PydanticAI 和底层的 OpenAI SDK 通常有日志功能。设置logging级别为DEBUG可以查看详细的请求和响应 JSON，这对于理解 LLM 的“思考”过程非常有帮助。

7.2 依赖注入失败或类型错误

问题：工具函数报错，提示某个依赖参数找不到（TypeError），或者运行时类型不匹配。
排查：
1. 参数名匹配：确保工具函数中接收依赖的参数名，与依赖函数返回值的“标识”完全一致。默认情况下，标识就是依赖函数的函数名。例如，依赖函数async def get_db()，工具函数参数就应该是db。
2. 显式指定依赖：你可以在@agent.tool装饰器或agent.tool()注册函数时，通过deps参数显式指定该工具需要哪些依赖，覆盖默认的命名匹配。例如：@agent.tool(deps=[get_db])。
3. 使用类型注解：为工具函数的依赖参数加上明确的类型注解（如db: DatabaseClient）。这虽然不影响运行时，但能让 mypy 等工具在静态检查时发现问题。

7.3 流式输出中断或不完整

问题：流式响应突然停止，前端收不到DONE事件，或者工具调用结果丢失。
排查：
1. 超时设置：检查你的 HTTP 服务器、反向代理（如 Nginx）以及 LLM API 客户端的超时设置。流式响应可能持续很长时间，需要将超时时间设置得足够长，或者禁用超时。
2. 异常捕获：如 6.3 节所述，务必在流式生成器的外层进行全面的异常捕获，并将错误信息 yield 出去。一个未捕获的异常会导致连接突然关闭。
3. 检查工具函数：确保你的工具函数是异步的（async def）且自身是健壮的。一个同步的、耗时的或会抛出异常的工具函数会阻塞整个流。
4. 网络稳定性：对于生产环境，考虑加入重试逻辑（特别是在调用外部 API 的工具中）和使用连接池。

7.4 性能优化建议

依赖缓存：如果某个依赖的获取成本很高（比如查询数据库），且在同一轮对话中多次运行 Agent 时不会改变，可以考虑使用@agent.dep装饰器并设置cache=True。PydanticAI 会在单次run或stream调用期间缓存该依赖的结果。
```
from pydantic_ai import Agent agent = Agent(...) @agent.dep(cache=True) # 这个依赖的结果会被缓存 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 的速率限制，防止滥用。