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是一个标记，用于告诉执行引擎：“这个参数不是普通输入，请在运行时为我解析并注入一个实例。” 依赖的来源可以是：

1. 函数（同步或异步）：框架会调用这个函数来获取值。
2. 类：框架会实例化这个类（也支持递归解析其构造函数依赖）。
3. 已解析的值：直接提供一个具体值。

from pydantic_ai import Agent, Depends import httpx async def get_http_client() -> httpx.AsyncClient: """依赖项函数：提供一个共享的 HTTP 客户端。""" client = httpx.AsyncClient(timeout=30.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.text

2.2.2 依赖解析流程与源码窥探

当执行fetch_data时，引擎不会期望调用者传递client参数。它会：

1. 识别出client参数有Depends(get_http_client)默认值。
2. 检查自身的“依赖容器”（一个维护了依赖项到其解析结果映射的上下文），看是否有get_http_client的缓存结果。
3. 如果没有，则执行get_http_client()函数。由于该函数是async generator（使用了yield），框架会以上下文管理器的方式运行它，进入try块，获取到yield出来的client对象，并将其提供给方法使用，同时缓存这个“正在运行”的生成器。
4. 方法执行完毕后，框架会继续执行生成器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('基于问题制定步骤', query=query) # 步骤2：执行。这会产生另一个子节点，并且依赖于步骤1的输出。 answer_result = await self.run('执行上述计划', plan=plan_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 提供了隐式但清晰的作用域控制。

1. 单例作用域（Singleton）：最常见的模式。依赖函数首次被解析后，其结果会被缓存，并在同一作用域（通常是整个Agent实例）的后续所有请求中重用。这对于数据库连接池、配置对象、模型客户端等重量级资源至关重要。通过使用yield模式的依赖，你还可以确保资源在使用完毕后被正确清理，即使中间发生了错误。

2. 请求/会话作用域（Request/Session）：在一次Agent.run()调用会话内保持唯一。例如，你可能会有一个依赖用于生成本次会话唯一的追踪 ID，或者维护本次多轮对话的上下文状态。PydanticAI 通过DependencyContext的不同层级来实现这一点。

3. 瞬态作用域（Transient）：每次需要时都重新创建。只需让依赖函数返回一个新实例即可，无需缓存。

实操心得：在设计依赖时，务必明确其生命周期。将本应是单例的依赖设计成瞬态，会导致性能瓶颈和资源泄露（如数据库连接数耗尽）。反之，将本应是会话级的依赖设计成单例，则会导致不同用户或请求间的状态污染。一个简单的判断方法是：这个依赖持有的资源或状态是否与特定的“一次运行”强相关？如果是，考虑会话作用域；如果与整个应用进程相关，考虑单例。

3.2 循环依赖与动态依赖的解决之道

在复杂的业务逻辑中，依赖关系可能形成循环（A 依赖 B，B 也依赖 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 节点类型与执行语义

在图引擎中，节点不仅仅是“一个函数调用”。根据其行为，可以分为几种类型：

1. 计算节点：纯函数或异步函数，不涉及 LLM 调用。例如，数据格式化、字符串处理、调用一个普通 API。由依赖注入系统提供输入，执行后输出结果。
2. LLM 调用节点：通过self.run创建。这是图的核心。引擎需要处理提示词模板渲染、模型客户端调用、响应解析（根据返回类型 Pydantic 模型）、Token 计数和费用计算等。
3. 控制流节点：这是图引擎的高级能力。虽然 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('分析以下文本的情感倾向', text=text) entities_future = self.run('提取以下文本中的命名实体', text=text) summary_future = self.run('总结以下文本', text=text) # `await` 会等待所有未来对象完成，但它们的执行可能是并发的。 sentiment, entities, summary = await asyncio.gather( sentiment_future, entities_future, summary_future ) return FullAnalysis( sentiment=sentiment.data, entities=entities.data, summary=summary.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(retries=3): def decorator(func): @tenacity.retry( stop=tenacity.stop_after_attempt(retries), retry=tenacity.retry_if_exception_type(RateLimitError), wait=tenacity.wait_exponential(multiplier=1, min=4, max=10) ) async def wrapper(*args, **kwargs): return await func(*args, **kwargs) return wrapper return decorator @retry_on_rate_limit(retries=5) 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(ge=0, le=1) # 检索请求 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(ge=0, le=1, 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_results=request.top_k ) docs = [ RetrievedDoc(content=doc, source=meta['source'], relevance_score=score) for doc, meta, score in zip(results['documents'][0], results['metadatas'][0], results['distances'][0]) ] return RetrievalResult(query=request.query, documents=docs) async def get_vector_store() -> VectorStoreDependency: # 使用yield模式管理生命周期 store = VectorStoreDependency() yield store # 如果需要清理，可以在这里进行 # await store.client.close() async def get_openai_client() -> AsyncOpenAI: client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY")) yield client

5.2 实现智能体与图执行

现在，我们创建智能体，将检索和生成两个步骤组织成一个工作流。

class RAGAgent(Agent): system_prompt = """你是一个专业的问答助手。请严格根据提供的参考文档来回答问题。 如果文档中包含答案，请用清晰的语言总结并引用文档[索引]。 如果文档中不包含答案，请直接说“根据现有资料无法回答此问题”。不要编造信息。""" def __init__(self): # 可以在这里注入一些默认依赖或配置 super().__init__(model='gpt-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(query=question, top_k=3) retrieval_result: RetrievalResult = await vector_store.search(retrieval_request) if not retrieval_result.documents: # 检索为空，直接返回 return AnswerWithCitations( answer="根据现有资料无法回答此问题。", citations=[], confidence=0.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( answer=answer_text, citations=citation_indices, confidence=min(1.0, len(citation_indices) / len(retrieval_result.documents)) # 简单置信度计算 )

5.3 运行与观察

现在，我们可以运行这个智能体，并观察双核引擎如何工作。

import asyncio async def main(): agent = RAGAgent() result = await agent.run(question="PydanticAI 的主要特点是什么？") print(result.data) # 输出类似： # AnswerWithCitations( # answer="PydanticAI 的主要特点是提供了类型安全的依赖注入系统和图执行引擎，用于构建可靠的AI工作流。[0]", # citations=[0], # confidence=0.333... # ) # 我们还可以深入查看运行细节 print(f"本次运行消耗的Token数: {result.usage.total_tokens}") print(f"运行步骤: {result.steps}") # 这里会展示图执行中的各个节点信息 if __name__ == "__main__": asyncio.run(main())

在这个流程中：

1. 依赖注入系统：当answer_question被调用时，框架解析出它需要vector_store和openai_client。它从依赖上下文中获取（或创建）这两个单例对象，并注入到方法中。
2. 图执行引擎：answer_question方法本身是一个节点。其内部，await vector_store.search(...)是一个同步调用（计算节点），而await self.run(prompt)是一个 LLM 调用节点。引擎理解这两个节点是顺序依赖关系（生成依赖检索结果），并按序执行。如果未来我们扩展智能体，让它并行检索多个不同的数据库，图引擎就能自动并发执行这些独立的检索节点。

6. 性能调优、调试与常见问题排查

使用如此高级的抽象，当出现问题或需要优化时，我们该如何下手？

6.1 性能分析与优化点

1. 依赖缓存检查：确保重量级依赖（如数据库连接、模型客户端）被正确缓存为单例。可以添加日志来确认依赖函数是否被重复调用。
2. 图并行化审视：检查你的@agent.act方法内部，是否存在可以并行执行的独立self.run或计算任务。将它们用asyncio.gather包装，以充分利用图引擎的并发能力。
3. 流式响应优化：如果使用流式，确保消费者（前端或下游）能够及时处理数据，避免生产者（LLM）被阻塞。考虑使用异步队列进行缓冲。
4. Token 与成本监控：AgentRun对象的usage属性详细记录了每次运行的 Token 消耗。对于复杂工作流，可以将其记录到监控系统，分析成本热点。

6.2 调试技巧与工具

1. 结构化日志：为你的依赖函数和 Agent 方法添加详细的日志。PydanticAI 本身也会在 DEBUG 级别记录依赖解析、图节点执行等关键事件。配置日志格式，包含run_id或step_id，便于追踪一次完整请求的链路。
2. 检查AgentRun对象：每次运行返回的result不仅包含数据，还有完整的steps列表。每个Step对象包含了该节点的输入、输出、状态、开始/结束时间以及可能发生的错误。这是事后调试的宝贵信息。
3. 可视化执行图（高级）：虽然 PydanticAI 未内置可视化工具，但你可以通过 hook 或监听器，在节点开始/结束时收集信息，然后使用graphviz等库生成本次运行的执行图，直观查看节点依赖和执行时长。
4. 使用pdb或ipdb：在依赖函数或@agent.act方法内部设置断点，是理解执行流程和排查逻辑错误的最直接方式。注意异步环境下的调试。

6.3 常见问题速查表

6.4 进阶：自定义代理与中间件

PydanticAI 的架构是开放的。你可以通过创建自定义的Agent子类或使用中间件（Middleware）来注入横切关注点逻辑，如：

• 日志记录中间件：记录每次run的输入、输出和耗时。
• 缓存中间件：对相同的提示词和参数进行缓存，减少 LLM 调用和成本。
• 限流中间件：控制并发请求数，防止超过模型提供商的速率限制。
• 监控中间件：将运行指标发送到 Prometheus 或 Datadog。

这通常通过重写Agent的_run或_run_stream方法，或在依赖链中插入代理对象来实现。这需要你对框架的源码有更深的理解，但也是发挥其最大威力的途径。