1. 项目概述从“数据校验”到“智能体编排”的范式跃迁如果你和我一样在过去几年里深度使用过 Pydantic那么你对它的印象很可能还停留在“那个非常好用的数据验证库”。确实从 API 请求参数校验到配置项管理Pydantic 凭借其基于 Python 类型注解的优雅设计几乎成了 Python 生态中数据验证的代名词。然而当我第一次深入 PydanticAI 的源码时我才意识到这个项目远不止是 Pydantic 在 AI 领域的简单套壳。它正在尝试解决一个更底层、也更棘手的问题如何为复杂、动态且充满不确定性的 AI 智能体工作流构建一套类型安全、可组合且高效执行的编程范式。简单来说PydanticAI 的核心目标是让你像定义 Pydantic 模型一样去定义和运行一个 AI 智能体或复杂工作流。你不再需要写一大堆胶水代码去手动拼接提示词、调用模型、解析输出、处理错误而是通过声明式的类型注解描述清楚每个步骤的输入、输出和依赖关系。PydanticAI 背后的“双核引擎”——类型安全的依赖注入系统和图执行引擎——会自动帮你搞定一切。这听起来有点抽象但你可以把它想象成从“手动组装流水线”到“声明式蓝图驱动自动化工厂”的转变。前者需要你亲自拧每一个螺丝后者你只需要画好设计图工厂就会自动调度资源、按序生产。这个转变的背后是当前 AI 应用开发中普遍存在的痛点代码脆弱、难以调试、组合性差。一个简单的智能体可能涉及模型调用、工具使用、条件分支和状态管理传统的命令式编程会让这些逻辑散落在各处一旦出错追踪起来如同大海捞针。PydanticAI 试图用“类型”这把利器为这片混沌之地引入秩序。接下来我们就潜入源码看看这两大核心引擎是如何协同工作将声明式的优雅转化为运行时的高效与可靠的。2. 核心架构总览声明式智能体的运行基石在拆解双核之前我们需要先理解 PydanticAI 定义智能体的基本单元Agent和Model。这和我们熟悉的 Pydantic 模型一脉相承但被赋予了新的使命。2.1 智能体即模型Agent类的本质在 PydanticAI 中一个智能体本质上是一个继承了pydantic_ai.Agent的类。这个类的类变量system_prompt定义了它的系统指令而它的方法需要用agent.act装饰则定义了它可以执行的动作或步骤。最精妙的设计在于这些方法的参数和返回值都通过 Python 的类型注解来定义并且返回值通常也是一个 Pydantic 的BaseModel子类。from pydantic import BaseModel from pydantic_ai import Agent class QueryResult(BaseModel): answer: str confidence: float class MyResearchAgent(Agent): system_prompt 你是一个研究助手。 agent.act async def search_and_summarize(self, topic: str) - QueryResult: # 这个方法体在“声明式”视角下甚至不是必需的。 # 执行逻辑由框架根据类型注解和装饰器注入。 ...看到这里你可能会有疑问search_and_summarize方法体里什么都没写它怎么工作这就是依赖注入和图执行引擎要解决的问题。方法签名(self, topic: str) - QueryResult本身就是一份完整的“契约”。它告诉框架“我需要一个topic字符串作为输入经过我的处理可能是调用大模型、使用工具等我会产出一个符合QueryResult结构的数据。” 框架的责任就是履行这份契约。2.2 依赖注入系统类型注解驱动的资源供给依赖注入Dependency Injection DI不是什么新概念但在动态类型语言 Python 中实现类型安全的 DI并用于 AI 工作流PydanticAI 的做法很有启发性。它的 DI 系统核心围绕Depends类和Agent的dependencies参数展开。2.2.1Depends声明你的依赖Depends是一个标记用于告诉执行引擎“这个参数不是普通输入请在运行时为我解析并注入一个实例。” 依赖的来源可以是函数同步或异步框架会调用这个函数来获取值。类框架会实例化这个类也支持递归解析其构造函数依赖。已解析的值直接提供一个具体值。from pydantic_ai import Agent, Depends import httpx async def get_http_client() - httpx.AsyncClient: 依赖项函数提供一个共享的 HTTP 客户端。 client httpx.AsyncClient(timeout30.0) try: yield client # 使用yield可实现类似FastAPI的上下文管理器依赖 finally: await client.aclose() class MyAgent(Agent): agent.act async def fetch_data( self, url: str, client: httpx.AsyncClient Depends(get_http_client) # 声明依赖 ) - str: response await client.get(url) return response.text2.2.2 依赖解析流程与源码窥探当执行fetch_data时引擎不会期望调用者传递client参数。它会识别出client参数有Depends(get_http_client)默认值。检查自身的“依赖容器”一个维护了依赖项到其解析结果映射的上下文看是否有get_http_client的缓存结果。如果没有则执行get_http_client()函数。由于该函数是async generator使用了yield框架会以上下文管理器的方式运行它进入try块获取到yield出来的client对象并将其提供给方法使用同时缓存这个“正在运行”的生成器。方法执行完毕后框架会继续执行生成器finally块中的清理代码await client.aclose()。注意这里的依赖缓存和作用域管理是 DI 系统的关键。PydanticAI 的依赖可以有单例整个 Agent 生命周期、会话一次运行等不同作用域这能有效避免重复创建昂贵资源如数据库连接、大模型客户端。在源码pydantic_ai/dependencies.py中Dependency类和DependencyContext类共同管理着这套复杂的生命周期。2.2.3 为什么是类型安全类型安全体现在两个层面。首先在编码时你的 IDE 可以通过类型注解提供准确的补全和错误检查。其次在运行时PydanticAI 会利用 Pydantic 的能力对注入的依赖值进行验证确保其符合声明的类型。如果get_http_client错误地返回了一个aiohttp.ClientSession框架在注入前就会抛出验证错误而不是等到方法内部调用时才出现AttributeError。这大大提升了框架的健壮性。2.3 图执行引擎将方法调用转化为工作流如果说依赖注入解决了“资源从哪里来”的问题那么图执行引擎Graph Execution Engine解决的就是“步骤如何执行”的问题。当你的智能体方法不仅仅是一个简单的函数而是可能包含对大模型的多次调用、条件逻辑、循环时一个线性的执行模型就不够用了。PydanticAI 将这些方法及其内部可能的子步骤建模为一个有向无环图。2.3.1 图的构建从装饰器到节点当你用agent.act装饰一个方法时你不仅仅定义了一个可调用对象更是定义了一个图节点的模板。这个节点的“运行逻辑”是在装饰时被分析和部分确定的。引擎会解析方法的签名、依赖、返回类型并将其注册为图中的一个可执行节点。更复杂的情况出现在方法体内部使用self.run或self.run_stream来调用模型时。这些调用本身也会成为图中的一个子节点。class ReasoningAgent(Agent): agent.act async def complex_task(self, query: str) - str: # 步骤1规划。这会产生一个子节点。 plan_result await self.run(基于问题制定步骤, queryquery) # 步骤2执行。这会产生另一个子节点并且依赖于步骤1的输出。 answer_result await self.run(执行上述计划, planplan_result.data) return answer_result.data在上面的例子中complex_task节点内部包含了两个顺序执行的子节点。图执行引擎会识别这种依赖关系answer_result依赖plan_result并据此安排执行顺序。2.3.2 调度与执行异步并发的艺术图执行引擎的核心优势在于其调度能力。对于没有依赖关系的节点引擎可以并发执行它们。例如如果你的智能体需要同时查询两个不同的 API 来获取信息这两个查询操作可以作为独立的节点被引擎调度到不同的异步任务中同时运行从而缩短整体响应时间。在源码pydantic_ai/runs/runner.py中AgentRun和GraphRunner类承担了主要的调度职责。它们维护着节点的状态等待、就绪、运行中、完成、失败并使用异步队列来管理可执行的任务。当一个节点的所有前置依赖都满足时它就被放入执行队列。2.3.3 流式支持与中间状态图执行引擎天然适合流式处理。当使用self.run_stream时它返回的是一个异步生成器。引擎会逐步产出每个“令牌”或中间结果。这对于构建实时响应的聊天应用或需要逐步展示思考过程的链式推理Chain-of-Thought场景至关重要。引擎需要细粒度地管理这些流式节点的状态确保数据流的正确传递和资源的及时清理。3. 类型安全依赖注入系统的深度解析理解了基本概念后我们深入依赖注入系统的几个关键设计细节这些细节决定了它的强大与灵活。3.1 依赖作用域与生命周期管理依赖项的生命周期管理是生产级应用必须考虑的问题。PydanticAI 提供了隐式但清晰的作用域控制。单例作用域Singleton最常见的模式。依赖函数首次被解析后其结果会被缓存并在同一作用域通常是整个Agent实例的后续所有请求中重用。这对于数据库连接池、配置对象、模型客户端等重量级资源至关重要。通过使用yield模式的依赖你还可以确保资源在使用完毕后被正确清理即使中间发生了错误。请求/会话作用域Request/Session在一次Agent.run()调用会话内保持唯一。例如你可能会有一个依赖用于生成本次会话唯一的追踪 ID或者维护本次多轮对话的上下文状态。PydanticAI 通过DependencyContext的不同层级来实现这一点。瞬态作用域Transient每次需要时都重新创建。只需让依赖函数返回一个新实例即可无需缓存。实操心得在设计依赖时务必明确其生命周期。将本应是单例的依赖设计成瞬态会导致性能瓶颈和资源泄露如数据库连接数耗尽。反之将本应是会话级的依赖设计成单例则会导致不同用户或请求间的状态污染。一个简单的判断方法是这个依赖持有的资源或状态是否与特定的“一次运行”强相关如果是考虑会话作用域如果与整个应用进程相关考虑单例。3.2 循环依赖与动态依赖的解决之道在复杂的业务逻辑中依赖关系可能形成循环A 依赖 BB 也依赖 A或者依赖本身需要根据运行时条件动态决定。PydanticAI 的 DI 系统对此有应对策略。循环依赖原生的基于构造函数的 DI 很难处理循环依赖。PydanticAI 主要采用“属性注入”或“方法注入”来规避。即不在__init__中注入而是在类的方法中通过Depends声明依赖。由于方法是在对象已创建后调用的因此打破了循环。在源码中依赖解析器会检测这种循环并抛出清晰的错误信息引导开发者重构。动态依赖有时依赖的具体实现需要在运行时根据配置、用户身份等决定。这可以通过“依赖工厂”模式实现。即创建一个高级别的依赖函数它内部根据条件返回不同的低级依赖实例。from pydantic import BaseModel from pydantic_ai import Depends class ModelConfig(BaseModel): provider: str # openai or anthropic async def get_llm_client(config: ModelConfig Depends(get_config)): if config.provider openai: from openai import AsyncOpenAI return AsyncOpenAI() else: from anthropic import AsyncAnthropic return AsyncAnthropic() class MyAgent(Agent): agent.act async def ask(self, question: str, llm_client Depends(get_llm_client)): # llm_client 会根据配置动态决定是 OpenAI 还是 Anthropic 客户端 ...这里的get_config本身也是一个依赖它可能从环境变量、数据库或请求上下文中读取配置。这种组合依赖的能力使得系统极具弹性。3.3 依赖验证与错误处理类型安全的核心保障在于验证。PydanticAI 在解析依赖后会利用 Pydantic 对结果进行验证确保其符合参数的类型注解。如果依赖函数返回None但参数类型是str框架会在调用业务方法前就抛出pydantic.ValidationError。这带来了巨大的调试便利性。错误被定位到了依赖解析阶段而不是业务逻辑深处。你可以清晰地看到是哪个依赖项没有满足契约。框架通常会提供详细的错误信息包括期望的类型、实际收到的值、以及出错的依赖路径。注意事项虽然框架提供了验证但依赖函数内部仍应做好基本的错误处理和资源管理。例如一个获取数据库连接的依赖如果连接失败应该抛出明确的异常如DependencyError的子类而不是返回None或一个无效对象让下游验证失败。这样错误信息会更具有业务语义。4. 图执行引擎的运作机制与高级特性图执行引擎是 PydanticAI 处理复杂工作流的大脑。让我们看看它是如何运作并支持一些高级模式的。4.1 节点类型与执行语义在图引擎中节点不仅仅是“一个函数调用”。根据其行为可以分为几种类型计算节点纯函数或异步函数不涉及 LLM 调用。例如数据格式化、字符串处理、调用一个普通 API。由依赖注入系统提供输入执行后输出结果。LLM 调用节点通过self.run创建。这是图的核心。引擎需要处理提示词模板渲染、模型客户端调用、响应解析根据返回类型 Pydantic 模型、Token 计数和费用计算等。控制流节点这是图引擎的高级能力。虽然 PydanticAI 主要采用声明式但它也支持通过if/else逻辑或基于运行结果的动态分支。这通常在agent.act方法内部通过 Python 原生控制流实现引擎会将其解释为图的条件边。更复杂的循环for,while理论上也可以通过节点和边的动态生成来模拟但这通常需要更精巧的设计有时直接在一个节点内用命令式逻辑处理反而更清晰。4.2 并行与聚合模式图引擎最直观的优化就是并行执行独立任务。假设一个智能体需要分析一篇长文的情感、提取关键实体并总结而这三个任务互不依赖。class AnalysisAgent(Agent): agent.act async def analyze_article(self, text: str) - FullAnalysis: # 这三个 self.run 调用没有相互依赖图引擎会尝试并行执行它们。 sentiment_future self.run(分析以下文本的情感倾向, texttext) entities_future self.run(提取以下文本中的命名实体, texttext) summary_future self.run(总结以下文本, texttext) # await 会等待所有未来对象完成但它们的执行可能是并发的。 sentiment, entities, summary await asyncio.gather( sentiment_future, entities_future, summary_future ) return FullAnalysis( sentimentsentiment.data, entitiesentities.data, summarysummary.data )引擎会识别出这三个run调用可以并行并创建对应的节点。asyncio.gather等待所有节点完成然后聚合结果。这比顺序执行快了近三倍。4.3 流式处理与实时交互对于self.run_stream图引擎的处理更为精细。它不会等待整个流完成才将节点标记为“完成”而是将流式响应本身作为一个可迭代的输出。下游节点如果依赖这个流式节点的结果则需要特殊处理例如等待流结束或者自己也以流式方式消费。这在实现打字机效果、实时进度更新或复杂的链式流式推理时非常有用。引擎需要确保流式数据在节点间的正确传递并管理好背压下游处理速度跟不上上游生产速度的情况。源码中的关键在pydantic_ai/runs/run.py中AgentRun对象管理着steps。每个Step对应图中的一个节点执行记录。对于流式运行Step会包含一个stream属性它是一个异步生成器。引擎的调度器需要能够协调这些流式步骤与其他常规步骤的执行。4.4 错误传播与重试机制在一个图中某个节点的失败如何处理PydanticAI 的图引擎通常采用“错误向上传播”的策略。如果一个节点执行失败如 LLM 调用超时、返回格式无法解析这个错误会使得该节点标记为失败。任何依赖该节点输出的后续节点会因为输入不满足而无法执行或直接标记为失败。更健壮的系统需要重试机制。PydanticAI 可以与外部重试库如tenacity结合或者在依赖层面实现重试逻辑。例如你可以创建一个retry_on_rate_limit的装饰器包装你的 LLM 客户端依赖函数使其在遇到速率限制错误时自动重试。import tenacity from openai import RateLimitError def retry_on_rate_limit(retries3): def decorator(func): tenacity.retry( stoptenacity.stop_after_attempt(retries), retrytenacity.retry_if_exception_type(RateLimitError), waittenacity.wait_exponential(multiplier1, min4, max10) ) async def wrapper(*args, **kwargs): return await func(*args, **kwargs) return wrapper return decorator retry_on_rate_limit(retries5) async def get_robust_llm_client(): return AsyncOpenAI() # 在Agent中使用这个带有重试的依赖将重试、熔断、降级等弹性模式实现在依赖层可以使你的业务逻辑图节点保持干净和专注。5. 双核协同实战构建一个类型安全的RAG智能体理论说得再多不如一个实例。让我们用 PydanticAI 的双核架构构建一个检索增强生成RAG智能体。这个智能体会接受用户问题从矢量数据库中检索相关文档然后让大模型基于这些文档生成答案。5.1 定义数据模型与依赖首先我们用 Pydantic 定义清晰的数据契约。from pydantic import BaseModel, Field from typing import List import numpy as np # 检索到的文档片段 class RetrievedDoc(BaseModel): content: str source: str relevance_score: float Field(ge0, le1) # 检索请求 class RetrievalRequest(BaseModel): query: str top_k: int 5 # 检索结果 class RetrievalResult(BaseModel): query: str documents: List[RetrievedDoc] # 最终的答案 class AnswerWithCitations(BaseModel): answer: str Field(description基于检索文档生成的最终答案) citations: List[int] Field(description引用的文档索引从0开始) confidence: float Field(ge0, le1, description答案置信度)接下来定义核心依赖矢量数据库连接器和嵌入模型。from pydantic_ai import Agent, Depends import httpx # 假设我们使用ChromaDB和OpenAI Embeddings import chromadb from openai import AsyncOpenAI class VectorStoreDependency: 矢量数据库依赖单例作用域 def __init__(self): self.client chromadb.PersistentClient(path./chroma_db) self.collection self.client.get_or_create_collection(knowledge_base) async def search(self, request: RetrievalRequest) - RetrievalResult: # 1. 将查询转换为向量这里简化实际需调用嵌入模型 # embedding await embed_model.embed(request.query) # 2. 在矢量数据库中搜索 results self.collection.query( query_embeddings[np.random.randn(1536)], # 模拟向量 n_resultsrequest.top_k ) docs [ RetrievedDoc(contentdoc, sourcemeta[source], relevance_scorescore) for doc, meta, score in zip(results[documents][0], results[metadatas][0], results[distances][0]) ] return RetrievalResult(queryrequest.query, documentsdocs) async def get_vector_store() - VectorStoreDependency: # 使用yield模式管理生命周期 store VectorStoreDependency() yield store # 如果需要清理可以在这里进行 # await store.client.close() async def get_openai_client() - AsyncOpenAI: client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) yield client5.2 实现智能体与图执行现在我们创建智能体将检索和生成两个步骤组织成一个工作流。class RAGAgent(Agent): system_prompt 你是一个专业的问答助手。请严格根据提供的参考文档来回答问题。 如果文档中包含答案请用清晰的语言总结并引用文档[索引]。 如果文档中不包含答案请直接说“根据现有资料无法回答此问题”。不要编造信息。 def __init__(self): # 可以在这里注入一些默认依赖或配置 super().__init__(modelgpt-4-turbo) agent.act async def answer_question( self, question: str, vector_store: VectorStoreDependency Depends(get_vector_store), openai_client: AsyncOpenAI Depends(get_openai_client) ) - AnswerWithCitations: 核心RAG流程检索 - 生成。 图引擎会将此方法识别为一个包含两个主要子节点的图。 # 节点1检索 retrieval_request RetrievalRequest(queryquestion, top_k3) retrieval_result: RetrievalResult await vector_store.search(retrieval_request) if not retrieval_result.documents: # 检索为空直接返回 return AnswerWithCitations( answer根据现有资料无法回答此问题。, citations[], confidence0.0 ) # 构建上下文 context_parts [] for i, doc in enumerate(retrieval_result.documents): context_parts.append(f[文档{i}] {doc.content}\n来源{doc.source}) context \n\n.join(context_parts) # 节点2基于上下文的生成 # 注意这里直接调用self.run它会利用图引擎和已注入的openai_client prompt f 参考文档 {context} 问题{question} 请根据以上文档回答问题。如果答案来自文档请在答案末尾用[文档索引]标注引用。 llm_result await self.run(prompt) # self.run会使用Agent的model和system_prompt # 解析LLM输出提取引用索引这里简化实际可用更复杂的解析或函数调用 # 假设LLM返回的文本中包含了类似[0][1]的引用标记。 import re answer_text llm_result.data citation_indices list(map(int, re.findall(r\[(\d)\], answer_text))) return AnswerWithCitations( answeranswer_text, citationscitation_indices, confidencemin(1.0, len(citation_indices) / len(retrieval_result.documents)) # 简单置信度计算 )5.3 运行与观察现在我们可以运行这个智能体并观察双核引擎如何工作。import asyncio async def main(): agent RAGAgent() result await agent.run(questionPydanticAI 的主要特点是什么) print(result.data) # 输出类似 # AnswerWithCitations( # answerPydanticAI 的主要特点是提供了类型安全的依赖注入系统和图执行引擎用于构建可靠的AI工作流。[0], # citations[0], # confidence0.333... # ) # 我们还可以深入查看运行细节 print(f本次运行消耗的Token数: {result.usage.total_tokens}) print(f运行步骤: {result.steps}) # 这里会展示图执行中的各个节点信息 if __name__ __main__: asyncio.run(main())在这个流程中依赖注入系统当answer_question被调用时框架解析出它需要vector_store和openai_client。它从依赖上下文中获取或创建这两个单例对象并注入到方法中。图执行引擎answer_question方法本身是一个节点。其内部await vector_store.search(...)是一个同步调用计算节点而await self.run(prompt)是一个 LLM 调用节点。引擎理解这两个节点是顺序依赖关系生成依赖检索结果并按序执行。如果未来我们扩展智能体让它并行检索多个不同的数据库图引擎就能自动并发执行这些独立的检索节点。6. 性能调优、调试与常见问题排查使用如此高级的抽象当出现问题或需要优化时我们该如何下手6.1 性能分析与优化点依赖缓存检查确保重量级依赖如数据库连接、模型客户端被正确缓存为单例。可以添加日志来确认依赖函数是否被重复调用。图并行化审视检查你的agent.act方法内部是否存在可以并行执行的独立self.run或计算任务。将它们用asyncio.gather包装以充分利用图引擎的并发能力。流式响应优化如果使用流式确保消费者前端或下游能够及时处理数据避免生产者LLM被阻塞。考虑使用异步队列进行缓冲。Token 与成本监控AgentRun对象的usage属性详细记录了每次运行的 Token 消耗。对于复杂工作流可以将其记录到监控系统分析成本热点。6.2 调试技巧与工具结构化日志为你的依赖函数和 Agent 方法添加详细的日志。PydanticAI 本身也会在 DEBUG 级别记录依赖解析、图节点执行等关键事件。配置日志格式包含run_id或step_id便于追踪一次完整请求的链路。检查AgentRun对象每次运行返回的result不仅包含数据还有完整的steps列表。每个Step对象包含了该节点的输入、输出、状态、开始/结束时间以及可能发生的错误。这是事后调试的宝贵信息。可视化执行图高级虽然 PydanticAI 未内置可视化工具但你可以通过 hook 或监听器在节点开始/结束时收集信息然后使用graphviz等库生成本次运行的执行图直观查看节点依赖和执行时长。使用pdb或ipdb在依赖函数或agent.act方法内部设置断点是理解执行流程和排查逻辑错误的最直接方式。注意异步环境下的调试。6.3 常见问题速查表问题现象可能原因排查步骤与解决方案依赖注入失败报DependencyError1. 依赖函数抛出未处理异常。2. 依赖返回值类型与参数声明不匹配。3. 存在无法解决的循环依赖。1. 检查依赖函数内部的错误处理。2. 确认依赖函数返回类型使用isinstance或 Pydantic 手动验证。3. 检查依赖关系图将循环依赖改为方法注入或服务定位器模式。self.run调用返回的结果无法解析为声明的返回类型1. LLM 返回的文本格式不符合 Pydantic 模型期望。2. 使用了run但返回类型不是BaseModel或基础类型。1. 强化你的提示词工程要求模型以指定格式如 JSON输出。使用result.raw_response查看原始响应进行调试。2. 对于复杂解析考虑使用 OpenAI 的函数调用或 PydanticAI 的结构化输出特性。图执行顺序不符合预期1. 节点间存在未声明的隐式依赖。2. 在异步函数中错误地使用了同步阻塞调用。1. 确保所有数据依赖都通过函数参数或self.run的返回值明确传递。2. 将同步 IO 操作如文件读写、某些网络请求改为异步版本或使用asyncio.to_thread在单独线程中执行避免阻塞事件循环。流式响应中断或速度慢1. 消费者处理速度慢造成背压。2. 网络连接不稳定。3. 模型提供商流式接口问题。1. 在消费者端使用异步缓冲队列。2. 增加网络超时和重试逻辑。3. 检查模型提供商的状态页并考虑使用更稳定的模型或配置。内存使用量随时间增长1. 依赖未正确清理导致资源泄露如数据库连接未关闭。2. 缓存了过大的对象如整个对话历史且未设置过期。1. 确保所有使用yield的依赖函数都有正确的finally清理块。2. 审查依赖缓存策略对于会话级数据考虑使用弱引用或定期清理。使用内存分析工具如tracemalloc定位泄漏点。6.4 进阶自定义代理与中间件PydanticAI 的架构是开放的。你可以通过创建自定义的Agent子类或使用中间件Middleware来注入横切关注点逻辑如日志记录中间件记录每次run的输入、输出和耗时。缓存中间件对相同的提示词和参数进行缓存减少 LLM 调用和成本。限流中间件控制并发请求数防止超过模型提供商的速率限制。监控中间件将运行指标发送到 Prometheus 或 Datadog。这通常通过重写Agent的_run或_run_stream方法或在依赖链中插入代理对象来实现。这需要你对框架的源码有更深的理解但也是发挥其最大威力的途径。 PydanticAI 双核引擎解析：类型安全依赖注入与图执行如何重塑 AI 智能体开发
1. 项目概述从“数据校验”到“智能体编排”的范式跃迁如果你和我一样在过去几年里深度使用过 Pydantic那么你对它的印象很可能还停留在“那个非常好用的数据验证库”。确实从 API 请求参数校验到配置项管理Pydantic 凭借其基于 Python 类型注解的优雅设计几乎成了 Python 生态中数据验证的代名词。然而当我第一次深入 PydanticAI 的源码时我才意识到这个项目远不止是 Pydantic 在 AI 领域的简单套壳。它正在尝试解决一个更底层、也更棘手的问题如何为复杂、动态且充满不确定性的 AI 智能体工作流构建一套类型安全、可组合且高效执行的编程范式。简单来说PydanticAI 的核心目标是让你像定义 Pydantic 模型一样去定义和运行一个 AI 智能体或复杂工作流。你不再需要写一大堆胶水代码去手动拼接提示词、调用模型、解析输出、处理错误而是通过声明式的类型注解描述清楚每个步骤的输入、输出和依赖关系。PydanticAI 背后的“双核引擎”——类型安全的依赖注入系统和图执行引擎——会自动帮你搞定一切。这听起来有点抽象但你可以把它想象成从“手动组装流水线”到“声明式蓝图驱动自动化工厂”的转变。前者需要你亲自拧每一个螺丝后者你只需要画好设计图工厂就会自动调度资源、按序生产。这个转变的背后是当前 AI 应用开发中普遍存在的痛点代码脆弱、难以调试、组合性差。一个简单的智能体可能涉及模型调用、工具使用、条件分支和状态管理传统的命令式编程会让这些逻辑散落在各处一旦出错追踪起来如同大海捞针。PydanticAI 试图用“类型”这把利器为这片混沌之地引入秩序。接下来我们就潜入源码看看这两大核心引擎是如何协同工作将声明式的优雅转化为运行时的高效与可靠的。2. 核心架构总览声明式智能体的运行基石在拆解双核之前我们需要先理解 PydanticAI 定义智能体的基本单元Agent和Model。这和我们熟悉的 Pydantic 模型一脉相承但被赋予了新的使命。2.1 智能体即模型Agent类的本质在 PydanticAI 中一个智能体本质上是一个继承了pydantic_ai.Agent的类。这个类的类变量system_prompt定义了它的系统指令而它的方法需要用agent.act装饰则定义了它可以执行的动作或步骤。最精妙的设计在于这些方法的参数和返回值都通过 Python 的类型注解来定义并且返回值通常也是一个 Pydantic 的BaseModel子类。from pydantic import BaseModel from pydantic_ai import Agent class QueryResult(BaseModel): answer: str confidence: float class MyResearchAgent(Agent): system_prompt 你是一个研究助手。 agent.act async def search_and_summarize(self, topic: str) - QueryResult: # 这个方法体在“声明式”视角下甚至不是必需的。 # 执行逻辑由框架根据类型注解和装饰器注入。 ...看到这里你可能会有疑问search_and_summarize方法体里什么都没写它怎么工作这就是依赖注入和图执行引擎要解决的问题。方法签名(self, topic: str) - QueryResult本身就是一份完整的“契约”。它告诉框架“我需要一个topic字符串作为输入经过我的处理可能是调用大模型、使用工具等我会产出一个符合QueryResult结构的数据。” 框架的责任就是履行这份契约。2.2 依赖注入系统类型注解驱动的资源供给依赖注入Dependency Injection DI不是什么新概念但在动态类型语言 Python 中实现类型安全的 DI并用于 AI 工作流PydanticAI 的做法很有启发性。它的 DI 系统核心围绕Depends类和Agent的dependencies参数展开。2.2.1Depends声明你的依赖Depends是一个标记用于告诉执行引擎“这个参数不是普通输入请在运行时为我解析并注入一个实例。” 依赖的来源可以是函数同步或异步框架会调用这个函数来获取值。类框架会实例化这个类也支持递归解析其构造函数依赖。已解析的值直接提供一个具体值。from pydantic_ai import Agent, Depends import httpx async def get_http_client() - httpx.AsyncClient: 依赖项函数提供一个共享的 HTTP 客户端。 client httpx.AsyncClient(timeout30.0) try: yield client # 使用yield可实现类似FastAPI的上下文管理器依赖 finally: await client.aclose() class MyAgent(Agent): agent.act async def fetch_data( self, url: str, client: httpx.AsyncClient Depends(get_http_client) # 声明依赖 ) - str: response await client.get(url) return response.text2.2.2 依赖解析流程与源码窥探当执行fetch_data时引擎不会期望调用者传递client参数。它会识别出client参数有Depends(get_http_client)默认值。检查自身的“依赖容器”一个维护了依赖项到其解析结果映射的上下文看是否有get_http_client的缓存结果。如果没有则执行get_http_client()函数。由于该函数是async generator使用了yield框架会以上下文管理器的方式运行它进入try块获取到yield出来的client对象并将其提供给方法使用同时缓存这个“正在运行”的生成器。方法执行完毕后框架会继续执行生成器finally块中的清理代码await client.aclose()。注意这里的依赖缓存和作用域管理是 DI 系统的关键。PydanticAI 的依赖可以有单例整个 Agent 生命周期、会话一次运行等不同作用域这能有效避免重复创建昂贵资源如数据库连接、大模型客户端。在源码pydantic_ai/dependencies.py中Dependency类和DependencyContext类共同管理着这套复杂的生命周期。2.2.3 为什么是类型安全类型安全体现在两个层面。首先在编码时你的 IDE 可以通过类型注解提供准确的补全和错误检查。其次在运行时PydanticAI 会利用 Pydantic 的能力对注入的依赖值进行验证确保其符合声明的类型。如果get_http_client错误地返回了一个aiohttp.ClientSession框架在注入前就会抛出验证错误而不是等到方法内部调用时才出现AttributeError。这大大提升了框架的健壮性。2.3 图执行引擎将方法调用转化为工作流如果说依赖注入解决了“资源从哪里来”的问题那么图执行引擎Graph Execution Engine解决的就是“步骤如何执行”的问题。当你的智能体方法不仅仅是一个简单的函数而是可能包含对大模型的多次调用、条件逻辑、循环时一个线性的执行模型就不够用了。PydanticAI 将这些方法及其内部可能的子步骤建模为一个有向无环图。2.3.1 图的构建从装饰器到节点当你用agent.act装饰一个方法时你不仅仅定义了一个可调用对象更是定义了一个图节点的模板。这个节点的“运行逻辑”是在装饰时被分析和部分确定的。引擎会解析方法的签名、依赖、返回类型并将其注册为图中的一个可执行节点。更复杂的情况出现在方法体内部使用self.run或self.run_stream来调用模型时。这些调用本身也会成为图中的一个子节点。class ReasoningAgent(Agent): agent.act async def complex_task(self, query: str) - str: # 步骤1规划。这会产生一个子节点。 plan_result await self.run(基于问题制定步骤, queryquery) # 步骤2执行。这会产生另一个子节点并且依赖于步骤1的输出。 answer_result await self.run(执行上述计划, planplan_result.data) return answer_result.data在上面的例子中complex_task节点内部包含了两个顺序执行的子节点。图执行引擎会识别这种依赖关系answer_result依赖plan_result并据此安排执行顺序。2.3.2 调度与执行异步并发的艺术图执行引擎的核心优势在于其调度能力。对于没有依赖关系的节点引擎可以并发执行它们。例如如果你的智能体需要同时查询两个不同的 API 来获取信息这两个查询操作可以作为独立的节点被引擎调度到不同的异步任务中同时运行从而缩短整体响应时间。在源码pydantic_ai/runs/runner.py中AgentRun和GraphRunner类承担了主要的调度职责。它们维护着节点的状态等待、就绪、运行中、完成、失败并使用异步队列来管理可执行的任务。当一个节点的所有前置依赖都满足时它就被放入执行队列。2.3.3 流式支持与中间状态图执行引擎天然适合流式处理。当使用self.run_stream时它返回的是一个异步生成器。引擎会逐步产出每个“令牌”或中间结果。这对于构建实时响应的聊天应用或需要逐步展示思考过程的链式推理Chain-of-Thought场景至关重要。引擎需要细粒度地管理这些流式节点的状态确保数据流的正确传递和资源的及时清理。3. 类型安全依赖注入系统的深度解析理解了基本概念后我们深入依赖注入系统的几个关键设计细节这些细节决定了它的强大与灵活。3.1 依赖作用域与生命周期管理依赖项的生命周期管理是生产级应用必须考虑的问题。PydanticAI 提供了隐式但清晰的作用域控制。单例作用域Singleton最常见的模式。依赖函数首次被解析后其结果会被缓存并在同一作用域通常是整个Agent实例的后续所有请求中重用。这对于数据库连接池、配置对象、模型客户端等重量级资源至关重要。通过使用yield模式的依赖你还可以确保资源在使用完毕后被正确清理即使中间发生了错误。请求/会话作用域Request/Session在一次Agent.run()调用会话内保持唯一。例如你可能会有一个依赖用于生成本次会话唯一的追踪 ID或者维护本次多轮对话的上下文状态。PydanticAI 通过DependencyContext的不同层级来实现这一点。瞬态作用域Transient每次需要时都重新创建。只需让依赖函数返回一个新实例即可无需缓存。实操心得在设计依赖时务必明确其生命周期。将本应是单例的依赖设计成瞬态会导致性能瓶颈和资源泄露如数据库连接数耗尽。反之将本应是会话级的依赖设计成单例则会导致不同用户或请求间的状态污染。一个简单的判断方法是这个依赖持有的资源或状态是否与特定的“一次运行”强相关如果是考虑会话作用域如果与整个应用进程相关考虑单例。3.2 循环依赖与动态依赖的解决之道在复杂的业务逻辑中依赖关系可能形成循环A 依赖 BB 也依赖 A或者依赖本身需要根据运行时条件动态决定。PydanticAI 的 DI 系统对此有应对策略。循环依赖原生的基于构造函数的 DI 很难处理循环依赖。PydanticAI 主要采用“属性注入”或“方法注入”来规避。即不在__init__中注入而是在类的方法中通过Depends声明依赖。由于方法是在对象已创建后调用的因此打破了循环。在源码中依赖解析器会检测这种循环并抛出清晰的错误信息引导开发者重构。动态依赖有时依赖的具体实现需要在运行时根据配置、用户身份等决定。这可以通过“依赖工厂”模式实现。即创建一个高级别的依赖函数它内部根据条件返回不同的低级依赖实例。from pydantic import BaseModel from pydantic_ai import Depends class ModelConfig(BaseModel): provider: str # openai or anthropic async def get_llm_client(config: ModelConfig Depends(get_config)): if config.provider openai: from openai import AsyncOpenAI return AsyncOpenAI() else: from anthropic import AsyncAnthropic return AsyncAnthropic() class MyAgent(Agent): agent.act async def ask(self, question: str, llm_client Depends(get_llm_client)): # llm_client 会根据配置动态决定是 OpenAI 还是 Anthropic 客户端 ...这里的get_config本身也是一个依赖它可能从环境变量、数据库或请求上下文中读取配置。这种组合依赖的能力使得系统极具弹性。3.3 依赖验证与错误处理类型安全的核心保障在于验证。PydanticAI 在解析依赖后会利用 Pydantic 对结果进行验证确保其符合参数的类型注解。如果依赖函数返回None但参数类型是str框架会在调用业务方法前就抛出pydantic.ValidationError。这带来了巨大的调试便利性。错误被定位到了依赖解析阶段而不是业务逻辑深处。你可以清晰地看到是哪个依赖项没有满足契约。框架通常会提供详细的错误信息包括期望的类型、实际收到的值、以及出错的依赖路径。注意事项虽然框架提供了验证但依赖函数内部仍应做好基本的错误处理和资源管理。例如一个获取数据库连接的依赖如果连接失败应该抛出明确的异常如DependencyError的子类而不是返回None或一个无效对象让下游验证失败。这样错误信息会更具有业务语义。4. 图执行引擎的运作机制与高级特性图执行引擎是 PydanticAI 处理复杂工作流的大脑。让我们看看它是如何运作并支持一些高级模式的。4.1 节点类型与执行语义在图引擎中节点不仅仅是“一个函数调用”。根据其行为可以分为几种类型计算节点纯函数或异步函数不涉及 LLM 调用。例如数据格式化、字符串处理、调用一个普通 API。由依赖注入系统提供输入执行后输出结果。LLM 调用节点通过self.run创建。这是图的核心。引擎需要处理提示词模板渲染、模型客户端调用、响应解析根据返回类型 Pydantic 模型、Token 计数和费用计算等。控制流节点这是图引擎的高级能力。虽然 PydanticAI 主要采用声明式但它也支持通过if/else逻辑或基于运行结果的动态分支。这通常在agent.act方法内部通过 Python 原生控制流实现引擎会将其解释为图的条件边。更复杂的循环for,while理论上也可以通过节点和边的动态生成来模拟但这通常需要更精巧的设计有时直接在一个节点内用命令式逻辑处理反而更清晰。4.2 并行与聚合模式图引擎最直观的优化就是并行执行独立任务。假设一个智能体需要分析一篇长文的情感、提取关键实体并总结而这三个任务互不依赖。class AnalysisAgent(Agent): agent.act async def analyze_article(self, text: str) - FullAnalysis: # 这三个 self.run 调用没有相互依赖图引擎会尝试并行执行它们。 sentiment_future self.run(分析以下文本的情感倾向, texttext) entities_future self.run(提取以下文本中的命名实体, texttext) summary_future self.run(总结以下文本, texttext) # await 会等待所有未来对象完成但它们的执行可能是并发的。 sentiment, entities, summary await asyncio.gather( sentiment_future, entities_future, summary_future ) return FullAnalysis( sentimentsentiment.data, entitiesentities.data, summarysummary.data )引擎会识别出这三个run调用可以并行并创建对应的节点。asyncio.gather等待所有节点完成然后聚合结果。这比顺序执行快了近三倍。4.3 流式处理与实时交互对于self.run_stream图引擎的处理更为精细。它不会等待整个流完成才将节点标记为“完成”而是将流式响应本身作为一个可迭代的输出。下游节点如果依赖这个流式节点的结果则需要特殊处理例如等待流结束或者自己也以流式方式消费。这在实现打字机效果、实时进度更新或复杂的链式流式推理时非常有用。引擎需要确保流式数据在节点间的正确传递并管理好背压下游处理速度跟不上上游生产速度的情况。源码中的关键在pydantic_ai/runs/run.py中AgentRun对象管理着steps。每个Step对应图中的一个节点执行记录。对于流式运行Step会包含一个stream属性它是一个异步生成器。引擎的调度器需要能够协调这些流式步骤与其他常规步骤的执行。4.4 错误传播与重试机制在一个图中某个节点的失败如何处理PydanticAI 的图引擎通常采用“错误向上传播”的策略。如果一个节点执行失败如 LLM 调用超时、返回格式无法解析这个错误会使得该节点标记为失败。任何依赖该节点输出的后续节点会因为输入不满足而无法执行或直接标记为失败。更健壮的系统需要重试机制。PydanticAI 可以与外部重试库如tenacity结合或者在依赖层面实现重试逻辑。例如你可以创建一个retry_on_rate_limit的装饰器包装你的 LLM 客户端依赖函数使其在遇到速率限制错误时自动重试。import tenacity from openai import RateLimitError def retry_on_rate_limit(retries3): def decorator(func): tenacity.retry( stoptenacity.stop_after_attempt(retries), retrytenacity.retry_if_exception_type(RateLimitError), waittenacity.wait_exponential(multiplier1, min4, max10) ) async def wrapper(*args, **kwargs): return await func(*args, **kwargs) return wrapper return decorator retry_on_rate_limit(retries5) async def get_robust_llm_client(): return AsyncOpenAI() # 在Agent中使用这个带有重试的依赖将重试、熔断、降级等弹性模式实现在依赖层可以使你的业务逻辑图节点保持干净和专注。5. 双核协同实战构建一个类型安全的RAG智能体理论说得再多不如一个实例。让我们用 PydanticAI 的双核架构构建一个检索增强生成RAG智能体。这个智能体会接受用户问题从矢量数据库中检索相关文档然后让大模型基于这些文档生成答案。5.1 定义数据模型与依赖首先我们用 Pydantic 定义清晰的数据契约。from pydantic import BaseModel, Field from typing import List import numpy as np # 检索到的文档片段 class RetrievedDoc(BaseModel): content: str source: str relevance_score: float Field(ge0, le1) # 检索请求 class RetrievalRequest(BaseModel): query: str top_k: int 5 # 检索结果 class RetrievalResult(BaseModel): query: str documents: List[RetrievedDoc] # 最终的答案 class AnswerWithCitations(BaseModel): answer: str Field(description基于检索文档生成的最终答案) citations: List[int] Field(description引用的文档索引从0开始) confidence: float Field(ge0, le1, description答案置信度)接下来定义核心依赖矢量数据库连接器和嵌入模型。from pydantic_ai import Agent, Depends import httpx # 假设我们使用ChromaDB和OpenAI Embeddings import chromadb from openai import AsyncOpenAI class VectorStoreDependency: 矢量数据库依赖单例作用域 def __init__(self): self.client chromadb.PersistentClient(path./chroma_db) self.collection self.client.get_or_create_collection(knowledge_base) async def search(self, request: RetrievalRequest) - RetrievalResult: # 1. 将查询转换为向量这里简化实际需调用嵌入模型 # embedding await embed_model.embed(request.query) # 2. 在矢量数据库中搜索 results self.collection.query( query_embeddings[np.random.randn(1536)], # 模拟向量 n_resultsrequest.top_k ) docs [ RetrievedDoc(contentdoc, sourcemeta[source], relevance_scorescore) for doc, meta, score in zip(results[documents][0], results[metadatas][0], results[distances][0]) ] return RetrievalResult(queryrequest.query, documentsdocs) async def get_vector_store() - VectorStoreDependency: # 使用yield模式管理生命周期 store VectorStoreDependency() yield store # 如果需要清理可以在这里进行 # await store.client.close() async def get_openai_client() - AsyncOpenAI: client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) yield client5.2 实现智能体与图执行现在我们创建智能体将检索和生成两个步骤组织成一个工作流。class RAGAgent(Agent): system_prompt 你是一个专业的问答助手。请严格根据提供的参考文档来回答问题。 如果文档中包含答案请用清晰的语言总结并引用文档[索引]。 如果文档中不包含答案请直接说“根据现有资料无法回答此问题”。不要编造信息。 def __init__(self): # 可以在这里注入一些默认依赖或配置 super().__init__(modelgpt-4-turbo) agent.act async def answer_question( self, question: str, vector_store: VectorStoreDependency Depends(get_vector_store), openai_client: AsyncOpenAI Depends(get_openai_client) ) - AnswerWithCitations: 核心RAG流程检索 - 生成。 图引擎会将此方法识别为一个包含两个主要子节点的图。 # 节点1检索 retrieval_request RetrievalRequest(queryquestion, top_k3) retrieval_result: RetrievalResult await vector_store.search(retrieval_request) if not retrieval_result.documents: # 检索为空直接返回 return AnswerWithCitations( answer根据现有资料无法回答此问题。, citations[], confidence0.0 ) # 构建上下文 context_parts [] for i, doc in enumerate(retrieval_result.documents): context_parts.append(f[文档{i}] {doc.content}\n来源{doc.source}) context \n\n.join(context_parts) # 节点2基于上下文的生成 # 注意这里直接调用self.run它会利用图引擎和已注入的openai_client prompt f 参考文档 {context} 问题{question} 请根据以上文档回答问题。如果答案来自文档请在答案末尾用[文档索引]标注引用。 llm_result await self.run(prompt) # self.run会使用Agent的model和system_prompt # 解析LLM输出提取引用索引这里简化实际可用更复杂的解析或函数调用 # 假设LLM返回的文本中包含了类似[0][1]的引用标记。 import re answer_text llm_result.data citation_indices list(map(int, re.findall(r\[(\d)\], answer_text))) return AnswerWithCitations( answeranswer_text, citationscitation_indices, confidencemin(1.0, len(citation_indices) / len(retrieval_result.documents)) # 简单置信度计算 )5.3 运行与观察现在我们可以运行这个智能体并观察双核引擎如何工作。import asyncio async def main(): agent RAGAgent() result await agent.run(questionPydanticAI 的主要特点是什么) print(result.data) # 输出类似 # AnswerWithCitations( # answerPydanticAI 的主要特点是提供了类型安全的依赖注入系统和图执行引擎用于构建可靠的AI工作流。[0], # citations[0], # confidence0.333... # ) # 我们还可以深入查看运行细节 print(f本次运行消耗的Token数: {result.usage.total_tokens}) print(f运行步骤: {result.steps}) # 这里会展示图执行中的各个节点信息 if __name__ __main__: asyncio.run(main())在这个流程中依赖注入系统当answer_question被调用时框架解析出它需要vector_store和openai_client。它从依赖上下文中获取或创建这两个单例对象并注入到方法中。图执行引擎answer_question方法本身是一个节点。其内部await vector_store.search(...)是一个同步调用计算节点而await self.run(prompt)是一个 LLM 调用节点。引擎理解这两个节点是顺序依赖关系生成依赖检索结果并按序执行。如果未来我们扩展智能体让它并行检索多个不同的数据库图引擎就能自动并发执行这些独立的检索节点。6. 性能调优、调试与常见问题排查使用如此高级的抽象当出现问题或需要优化时我们该如何下手6.1 性能分析与优化点依赖缓存检查确保重量级依赖如数据库连接、模型客户端被正确缓存为单例。可以添加日志来确认依赖函数是否被重复调用。图并行化审视检查你的agent.act方法内部是否存在可以并行执行的独立self.run或计算任务。将它们用asyncio.gather包装以充分利用图引擎的并发能力。流式响应优化如果使用流式确保消费者前端或下游能够及时处理数据避免生产者LLM被阻塞。考虑使用异步队列进行缓冲。Token 与成本监控AgentRun对象的usage属性详细记录了每次运行的 Token 消耗。对于复杂工作流可以将其记录到监控系统分析成本热点。6.2 调试技巧与工具结构化日志为你的依赖函数和 Agent 方法添加详细的日志。PydanticAI 本身也会在 DEBUG 级别记录依赖解析、图节点执行等关键事件。配置日志格式包含run_id或step_id便于追踪一次完整请求的链路。检查AgentRun对象每次运行返回的result不仅包含数据还有完整的steps列表。每个Step对象包含了该节点的输入、输出、状态、开始/结束时间以及可能发生的错误。这是事后调试的宝贵信息。可视化执行图高级虽然 PydanticAI 未内置可视化工具但你可以通过 hook 或监听器在节点开始/结束时收集信息然后使用graphviz等库生成本次运行的执行图直观查看节点依赖和执行时长。使用pdb或ipdb在依赖函数或agent.act方法内部设置断点是理解执行流程和排查逻辑错误的最直接方式。注意异步环境下的调试。6.3 常见问题速查表问题现象可能原因排查步骤与解决方案依赖注入失败报DependencyError1. 依赖函数抛出未处理异常。2. 依赖返回值类型与参数声明不匹配。3. 存在无法解决的循环依赖。1. 检查依赖函数内部的错误处理。2. 确认依赖函数返回类型使用isinstance或 Pydantic 手动验证。3. 检查依赖关系图将循环依赖改为方法注入或服务定位器模式。self.run调用返回的结果无法解析为声明的返回类型1. LLM 返回的文本格式不符合 Pydantic 模型期望。2. 使用了run但返回类型不是BaseModel或基础类型。1. 强化你的提示词工程要求模型以指定格式如 JSON输出。使用result.raw_response查看原始响应进行调试。2. 对于复杂解析考虑使用 OpenAI 的函数调用或 PydanticAI 的结构化输出特性。图执行顺序不符合预期1. 节点间存在未声明的隐式依赖。2. 在异步函数中错误地使用了同步阻塞调用。1. 确保所有数据依赖都通过函数参数或self.run的返回值明确传递。2. 将同步 IO 操作如文件读写、某些网络请求改为异步版本或使用asyncio.to_thread在单独线程中执行避免阻塞事件循环。流式响应中断或速度慢1. 消费者处理速度慢造成背压。2. 网络连接不稳定。3. 模型提供商流式接口问题。1. 在消费者端使用异步缓冲队列。2. 增加网络超时和重试逻辑。3. 检查模型提供商的状态页并考虑使用更稳定的模型或配置。内存使用量随时间增长1. 依赖未正确清理导致资源泄露如数据库连接未关闭。2. 缓存了过大的对象如整个对话历史且未设置过期。1. 确保所有使用yield的依赖函数都有正确的finally清理块。2. 审查依赖缓存策略对于会话级数据考虑使用弱引用或定期清理。使用内存分析工具如tracemalloc定位泄漏点。6.4 进阶自定义代理与中间件PydanticAI 的架构是开放的。你可以通过创建自定义的Agent子类或使用中间件Middleware来注入横切关注点逻辑如日志记录中间件记录每次run的输入、输出和耗时。缓存中间件对相同的提示词和参数进行缓存减少 LLM 调用和成本。限流中间件控制并发请求数防止超过模型提供商的速率限制。监控中间件将运行指标发送到 Prometheus 或 Datadog。这通常通过重写Agent的_run或_run_stream方法或在依赖链中插入代理对象来实现。这需要你对框架的源码有更深的理解但也是发挥其最大威力的途径。