Datapizza AI：构建可靠、可观测、供应商无关的生成式AI应用框架-平芜编程栈

1. 项目概述：一个为工程师而生的AI应用框架

如果你和我一样，在过去一两年里折腾过各种大语言模型应用，从简单的聊天机器人到复杂的多智能体系统，那你一定深有体会：从原型到生产，这条路有多难走。我们常常陷在几个泥潭里：要么是框架抽象过度，出了问题像在黑盒里摸象，调试起来一头雾水；要么是供应商绑定太死，想换个模型试试，结果发现要重写半壁江山；更别提那些在测试环境跑得好好的，一上线就各种幺蛾子的“玄学”问题了。

今天要聊的这个Datapizza AI，就是冲着解决这些痛点来的。我第一次在GitHub上看到它时，就被它的Slogan吸引了——“Build reliable Gen AI solutions without overhead”。没有花里胡哨的过度封装，没有为了炫技而设计的复杂概念，它就是一个用Python写的、为速度而生的、无废话的GenAI框架。它的目标很明确：让你能快速、可靠地把你的AI智能体从开发环境部署到生产环境。

简单来说，Datapizza AI是一个API优先、可组合、可观测且供应商无关的框架。它不是一个试图包办一切的“全家桶”，而更像一套精心设计的乐高积木。它提供了构建AI应用所需的核心组件——智能体、工具、文档处理、向量存储、流水线——并且这些组件之间通过清晰、一致的接口连接。你可以用这些积木快速搭建起一个原型，更重要的是，当你想深入内部看看某个部件是怎么工作的，或者想替换掉某个供应商时，你会发现一切都清晰可控。

这个框架特别适合谁呢？我认为是三类人：一是需要快速构建生产级AI应用的工程师，它开箱即用的可观测性和模块化设计能省下大量调试和集成的时间；二是正在从其他框架（比如LangChain）迁移的团队，它的设计理念强调清晰的接口和更少的“魔法”，迁移成本相对较低；三是任何对AI应用的可控性和可调试性有高要求的开发者，它的“可观测性优先”设计理念，让整个AI工作流的每一步都变得透明。

2. 核心设计哲学：为什么是“无废话”框架？

在深入代码之前，我想先拆解一下Datapizza AI背后的设计哲学。这能帮你理解它为什么这样设计，以及它是否契合你的需求。我总结下来，主要是四个核心原则。

2.1 原则一：API优先，控制权归开发者

很多AI框架为了追求易用性，做了大量的抽象和封装。这在小项目里是福音，但在复杂的生产系统中，过度的抽象就成了灾难。当调用链出错时，你看到的可能只是一个模糊的“调用失败”信息，至于是在模型调用、工具执行还是上下文管理环节出的问题，你得像侦探一样去层层推断。

Datapizza AI选择了另一条路：API优先。这意味着框架的核心是一组定义清晰、行为可预测的接口（API）。例如，所有的AI客户端（OpenAI、Gemini等）都遵循相同的调用模式，所有的工具都有统一的装饰器定义方式。这种设计带来的最大好处是控制感和可预测性。你清楚地知道数据是如何在组件间流动的，当出现异常时，你能快速定位到是哪个接口的输入或输出不符合预期。

这种设计也体现在它的“多供应商支持”上。它并非通过一个统一的、高度抽象的“LLM”类来封装所有供应商，而是为每个主流供应商（OpenAI、Google Gemini、Anthropic、Mistral、Azure OpenAI）提供了独立的客户端实现。这些客户端都实现了相同的核心接口，因此你可以在不改变业务逻辑的情况下，通过更换一个客户端实例就切换模型供应商。这比那种在配置文件里改个模型名字，但底层却可能因为参数细微差别而报错的“伪抽象”要可靠得多。

2.2 原则二：可组合性，像搭积木一样构建系统

现代AI应用很少是单一模型的一次调用。它可能是检索增强生成（RAG），需要先解析文档、分块、嵌入、检索，再生成；也可能是多智能体协作，一个规划智能体需要调用搜索智能体和天气智能体。Datapizza AI通过可组合性来应对这种复杂性。

框架将常见功能封装成独立的、可复用的“模块”（Module）。比如，DoclingParser是一个文档解析模块，NodeSplitter是一个文本分块模块，OpenAIEmbedder是一个嵌入模块。你可以通过IngestionPipeline或DagPipeline将这些模块像管道一样连接起来，形成一个完整的工作流。

这种设计的美妙之处在于灵活性和可测试性。每个模块都可以独立开发和测试。如果你想尝试不同的分块策略，只需换掉NodeSplitter模块，而不影响前后的解析和嵌入逻辑。框架提供的DagPipeline更是支持定义有向无环图，让你可以构建非线性的、复杂的工作流，比如在检索后加入一个重排序（Reranking）步骤来提升精度。

2.3 原则三：可观测性内建，告别“黑盒”调试

这是Datapizza AI让我眼前一亮的特性。它把可观测性作为一等公民来设计，而不是事后补救的补丁。框架内置了基于OpenTelemetry的分布式追踪。这意味着，从你发起一个AI智能体调用开始，到它调用工具、访问记忆、生成回复的每一个子步骤，都可以被追踪并记录为一个独立的“跨度”（Span）。

想象一下这个场景：你的智能体回复了一个错误答案。在传统框架里，你可能需要到处加日志，猜测是检索出了问题还是模型理解有误。而在Datapizza AI中，你可以直接打开追踪视图，看到一个清晰的调用链：agent.run->tool.search->llm.generate。每个步骤的耗时、输入、输出（如果开启I/O追踪）、以及消耗的Token数都一目了然。这极大地加速了调试和性能剖析的过程。

更关键的是，这种追踪是标准化的（基于OpenTelemetry），这意味着你可以将追踪数据导出到Jaeger、Zipkin等你熟悉的观测平台，与你现有的微服务监控体系集成。

2.4 原则四：供应商无关，避免被单一平台锁死

AI领域变化日新月异，新的模型和供应商不断涌现。今天你可能用GPT-4，明天可能就想试试Claude 3.5或者DeepSeek。一个设计良好的框架应该让你能轻松地拥抱这种变化，而不是被锁死。

Datapizza AI通过清晰的接口和模块化设计实现了供应商无关性。它的核心组件（如客户端、嵌入器）都定义了抽象的基类。不同的供应商实现这些基类。因此，将OpenAI客户端换成Anthropic客户端，通常只需要修改一行初始化代码。你的业务逻辑——比如如何构建提示词、如何处理工具调用结果——完全不需要改动。

这种设计也延伸到了整个生态系统。无论是向量数据库（目前支持Qdrant）、重排序器（支持Cohere、Together AI）还是文档解析器（支持Azure AI Document Intelligence、Docling），框架都通过统一的接口进行集成，让你可以根据性能、成本和功能需求自由选型。

3. 从零开始：环境搭建与核心概念上手

理论说了不少，现在我们来点实际的。我会带你从安装开始，一步步搭建环境，并亲手运行几个核心示例，让你切身感受一下这个框架的“手感”。

3.1 安装与初始配置

安装非常简单，就像任何其他Python包一样。我建议先创建一个干净的虚拟环境，避免依赖冲突。

# 创建并激活虚拟环境（以conda为例） conda create -n datapizza-demo python=3.10 conda activate datapizza-demo # 安装核心框架 pip install datapizza-ai

如果你打算使用特定的AI供应商，可以安装对应的扩展包。这不是必须的，核心框架包含了基础接口，但扩展包提供了官方集成的便利性。

# 按需安装供应商客户端（可选） pip install datapizza-ai-clients-openai # pip install datapizza-ai-clients-google # pip install datapizza-ai-clients-anthropic

安装完成后，你需要准备好AI模型的API密钥。以OpenAI为例，你需要一个有效的OPENAI_API_KEY。你可以将它设置为环境变量，这是最安全也是推荐的方式。

# 在Linux/macOS的终端中 export OPENAI_API_KEY='your-api-key-here' # 在Windows的PowerShell中 $env:OPENAI_API_KEY='your-api-key-here'

注意：永远不要将API密钥硬编码在源代码中并提交到版本控制系统（如Git）。使用环境变量或安全的密钥管理服务（如AWS Secrets Manager, HashiCorp Vault）是生产环境的基本要求。

3.2 第一个智能体：与模型对话

让我们从一个最简单的“Hello World”开始：初始化一个客户端，然后向模型问好。

from datapizza.clients.openai import OpenAIClient # 初始化客户端。如果没有设置环境变量，也可以直接传入api_key参数（仅用于测试） client = OpenAIClient() # 会自动读取 OPENAI_API_KEY 环境变量 # 或者 client = OpenAIClient(api_key="sk-...") # 调用模型 result = client.invoke("Hi, how are you?") print(result.text)

运行这段代码，你应该会收到模型的一句问候回复。这看起来平平无奇，但请注意client.invoke返回的result对象。它不仅仅包含文本，根据你使用的客户端，它可能还包含原始响应、使用量（Token数）等元数据。这种一致的返回结构是API优先设计的一个体现。

3.3 赋予智能体“工具”：让AI能动手做事

单纯的对话能力有限，真正的智能体需要能调用外部工具来获取信息或执行操作。Datapizza AI让定义和使用工具变得非常简单。

from datapizza.agents import Agent from datapizza.clients.openai import OpenAIClient from datapizza.tools import tool # 1. 使用 @tool 装饰器定义一个工具函数 @tool def get_weather(city: str) -> str: """获取指定城市的天气信息。""" # 这里是一个模拟实现。真实场景中，你会在这里调用天气API。 # 函数必须有类型注解和文档字符串，这有助于AI理解如何使用它。 return f"The weather in {city} is sunny and 25°C." # 2. 创建客户端和智能体，并将工具传递给它 client = OpenAIClient() agent = Agent(name="my_assistant", client=client, tools=[get_weather]) # 3. 运行智能体。它会自动判断是否需要以及如何调用工具。 response = agent.run("What is the weather in Rome?") print(response.text) # 输出: The weather in Rome is sunny and 25°C.

这里发生了什么？当你调用agent.run时，框架内部会：

将你的问题、系统提示（如果有）和可用的工具描述一起发送给模型。
模型判断需要调用get_weather工具，并生成一个符合工具参数格式的调用请求。
框架执行这个工具调用（即运行get_weather("Rome")），得到结果。
框架将工具执行结果和原始问题再次发送给模型，让模型生成最终的自然语言回复。

整个过程对开发者是透明的。你只需要定义好工具函数，剩下的交给框架。这种设计模式非常清晰，也易于调试，因为你可以通过后面会讲到的追踪功能，清晰地看到“模型决定调用工具 -> 执行工具 -> 模型整合结果”的完整链路。

3.4 使用真实工具：连接外部世界

模拟工具不过瘾，我们试试连接真实世界的工具。Datapizza AI社区提供了一些预构建的工具，比如网络搜索。

# 首先安装DuckDuckGo搜索工具包 pip install datapizza-ai-tools-duckduckgo

from datapizza.agents import Agent from datapizza.clients.openai import OpenAIClient from datapizza.tools.duckduckgo import DuckDuckGoSearchTool client = OpenAIClient() # DuckDuckGoSearchTool 是一个已经定义好的工具类 agent = Agent(name="research_assistant", client=client, tools=[DuckDuckGoSearchTool()]) response = agent.run("What are the latest developments in quantum computing as of 2024?") print(response.text)

现在你的智能体已经可以联网搜索最新信息了！DuckDuckGoSearchTool封装了搜索的细节，你无需处理HTTP请求和HTML解析。这就是使用一个成熟框架的好处——站在巨人的肩膀上，快速集成成熟组件。

4. 深入核心：构建可观测与可协作的AI系统

掌握了基础，我们可以探索更强大的功能了。Datapizza AI的真正威力在于构建复杂的、可观测的、多智能体协作的系统。

4.1 开启上帝视角：详尽的执行追踪

还记得我们强调的可观测性吗？让我们看看它具体怎么用。框架的追踪功能默认可能是关闭或简化的，为了获得最详细的洞察，我们需要显式地开启它。

from datapizza.agents import Agent from datapizza.clients.openai import OpenAIClient from datapizza.tools.duckduckgo import DuckDuckGoSearchTool from datapizza.tracing import ContextTracing client = OpenAIClient() agent = Agent(name="assistant", client=client, tools=[DuckDuckGoSearchTool()]) # 使用 ContextTracing 上下文管理器来追踪一个操作 with ContextTracing().trace("my_ai_operation"): response = agent.run("Tell me some news about Bitcoin") # 在控制台，你会看到类似下面的追踪摘要输出

执行后，你会在终端看到一个结构化的追踪摘要表格。这个表格不仅告诉你总耗时和Span数量，更重要的是，它列出了每次模型调用的详细信息：

Model: 使用的是哪个模型（如gpt-4o-mini）。
Prompt Tokens: 提示词消耗的Token数。
Completion Tokens: 回复消耗的Token数。
Cached Tokens: 是否使用了缓存（如果模型支持）。

这对于成本监控和性能优化至关重要。你可以一眼看出是哪个步骤消耗了最多的Token或时间。如果部署了像Jaeger这样的可视化工具，你还能看到完整的调用链火焰图，精确到每个函数调用。

实操心得：在开发阶段，我强烈建议始终开启追踪。它就像给你的AI应用装上了调试器。很多逻辑错误和性能瓶颈，在清晰的调用链面前都无所遁形。生产环境中，你可以根据采样率有选择地记录追踪数据，以平衡洞察力和开销。

4.2 组建AI团队：实现多智能体协作

单个智能体能力有限，但让多个各有所长的智能体协作，就能解决复杂问题。Datapizza AI的Agent类原生支持这种“召唤”机制。

让我们构建一个旅行规划系统，它包含三个智能体：

天气专家：负责提供天气信息。
网络搜索专家：负责搜索网络上的景点、活动信息。
规划师：核心协调者，负责理解用户需求，并调用前两个专家来制定详细计划。

from datapizza.agents.agent import Agent from datapizza.clients.openai import OpenAIClient from datapizza.tools import tool from datapizza.tools.duckduckgo import DuckDuckGoSearchTool client = OpenAIClient(api_key="YOUR_API_KEY", model="gpt-4") # 1. 定义天气工具和天气专家智能体 @tool def get_weather(city: str) -> str: # 模拟数据，真实项目请接入API return f"""Forecast for {city} next week: - Monday: Sunny, 22°C - Tuesday: Partly cloudy, 20°C - Wednesday: Light rain, 18°C - Thursday: Sunny, 24°C - Friday: Clear, 23°C Perfect for hiking most days, but pack a raincoat for Wednesday.""" weather_agent = Agent( name="weather_expert", client=client, system_prompt="You are a weather expert. Provide detailed, accurate weather forecasts and outdoor activity recommendations based on the weather.", tools=[get_weather] ) # 2. 创建网络搜索专家智能体 web_search_agent = Agent( name="web_search_expert", client=client, system_prompt="You are a web search expert. Find and summarize relevant, up-to-date information from the internet about travel destinations, attractions, hiking trails, and local tips.", tools=[DuckDuckGoSearchTool()] ) # 3. 创建规划师智能体 planner_agent = Agent( name="planner", client=client, system_prompt="""You are a professional trip planner. Your task is to create detailed, practical, and enjoyable itineraries based on user requests. You have access to a weather expert and a web search expert. You MUST call them when you need weather forecasts or up-to-date information about places and activities. Synthesize their inputs into a coherent, day-by-day plan with recommendations on what to do, when to do it, and what to prepare.""" ) # 4. 关键一步：赋予规划师调用其他智能体的能力 planner_agent.can_call([weather_agent, web_search_agent]) # 5. 运行规划师 response = planner_agent.run( "I need to plan a 3-day hiking trip in the Seattle area next week. I love waterfalls and dense forests. Please suggest a detailed itinerary." ) print(response.text)

当你运行这段代码时，planner_agent会自主工作：

它首先分析用户请求，意识到需要天气和景点信息。
它会“召唤”weather_agent，询问“Seattle next week”的天气。
它会“召唤”web_search_agent，搜索“best waterfalls and forests hiking trails near Seattle”。
它收到两个专家的回复后，综合这些信息，生成一份包含日期、地点、活动建议、装备提示的详细旅行计划。

这种设计模式将复杂任务分解，让每个智能体专注于自己擅长的领域，通过清晰的协作协议（can_call）连接起来，极大地提升了系统的能力和可靠性。

5. 构建知识大脑：从文档处理到智能检索（RAG）

对于企业级应用，让AI基于私有知识库进行回答（RAG）是核心场景。Datapizza AI为构建RAG系统提供了一套完整的、模块化的流水线。

5.1 文档摄取流水线：将知识存入向量库

RAG的第一步是处理文档并创建可检索的索引。我们用一个完整的例子来演示如何解析PDF、分块、生成嵌入并存储到Qdrant向量数据库。

# 安装必要的依赖 pip install datapizza-ai-parsers-docling # 强大的文档解析器 pip install qdrant-client # Qdrant向量数据库客户端

from datapizza.core.vectorstore import VectorConfig from datapizza.embedders import ChunkEmbedder from datapizza.embedders.openai import OpenAIEmbedder from datapizza.modules.parsers.docling import DoclingParser from datapizza.modules.splitters import NodeSplitter from datapizza.pipeline import IngestionPipeline from datapizza.vectorstores.qdrant import QdrantVectorstore # 1. 初始化向量存储（使用内存模式，方便演示） vectorstore = QdrantVectorstore(location=":memory:") # 2. 初始化嵌入器 embedder = ChunkEmbedder(client=OpenAIEmbedder(model_name="text-embedding-3-small")) # 注意：text-embedding-3-small 的维度是1536 embedding_dimensions = 1536 # 3. 在向量库中创建集合（类似数据库的表），并定义向量字段的配置 vectorstore.create_collection( "company_handbook", vector_config=[VectorConfig(name="embedding", dimensions=embedding_dimensions)] ) # 4. 构建摄取流水线 pipeline = IngestionPipeline( modules=[ DoclingParser(), # 模块1: 解析PDF/DOCX，提取文本和结构 NodeSplitter(max_char=1024), # 模块2: 将长文本分割成适合的小块（chunks） embedder, # 模块3: 为每个文本块生成向量嵌入 ], vector_store=vectorstore, # 目标存储 collection_name="company_handbook" # 存储到的集合名 ) # 5. 运行流水线，处理文档 # 假设你有一个名为 `employee_handbook.pdf` 的文件 pipeline.run("path/to/your/employee_handbook.pdf") print("文档摄取完成！")

这个流水线清晰地展示了可组合性。IngestionPipeline按顺序执行三个模块。如果你想换用不同的分块策略（比如按句子分块），只需将NodeSplitter替换成其他Splitter模块。如果你想用Cohere的嵌入模型，只需将OpenAIEmbedder替换为CohereEmbedder。这种插拔式的设计让实验和迭代变得非常高效。

5.2 构建端到端RAG查询管道

知识库建好了，接下来就要构建查询流程。这通常包括：查询重写 -> 查询嵌入 -> 向量检索 -> 提示词构建 -> 答案生成。Datapizza AI的DagPipeline非常适合描述这种有向无环图式的工作流。

from datapizza.clients.openai import OpenAIClient from datapizza.embedders.openai import OpenAIEmbedder from datapizza.modules.prompt import ChatPromptTemplate from datapizza.modules.rewriters import ToolRewriter from datapizza.pipeline import DagPipeline from datapizza.vectorstores.qdrant import QdrantVectorstore # 初始化各组件 openai_client = OpenAIClient(model="gpt-4o-mini") embedder = OpenAIEmbedder(model_name="text-embedding-3-small") # 连接到正在运行的Qdrant实例（假设在本地6333端口） retriever = QdrantVectorstore(host="localhost", port=6333).as_retriever(collection_name="company_handbook", k=5) # 1. 创建DAG管道 dag_pipeline = DagPipeline() # 2. 添加各个处理节点（模块） dag_pipeline.add_module( "rewriter", ToolRewriter( client=openai_client, system_prompt="Rewrite the user's query to be more effective for retrieving relevant document chunks. Expand acronyms, clarify ambiguous terms." ) ) dag_pipeline.add_module("embedder", embedder) dag_pipeline.add_module("retriever", retriever) dag_pipeline.add_module( "prompt", ChatPromptTemplate( user_prompt_template="用户问题：{{user_prompt}}\n\n请根据以下相关资料进行回答：", retrieval_prompt_template="相关材料：\n{% for chunk in chunks %}[材料{{ loop.index }}] {{ chunk.text }}\n{% endfor %}" ) ) dag_pipeline.add_module("generator", openai_client) # 3. 定义节点之间的连接关系（数据流） dag_pipeline.connect("rewriter", "embedder", target_key="text") # 重写后的文本 -> 嵌入器 dag_pipeline.connect("embedder", "retriever", target_key="query_vector") # 嵌入向量 -> 检索器 dag_pipeline.connect("retriever", "prompt", target_key="chunks") # 检索到的块 -> 提示词模板 dag_pipeline.connect("prompt", "generator", target_key="memory") # 构建好的提示词 -> 生成器 # 4. 运行管道 query = "我们公司的年假政策是怎样的？" result = dag_pipeline.run({ "rewriter": {"user_prompt": query}, "prompt": {"user_prompt": query}, # 原始问题也传给提示词模板 "retriever": {"collection_name": "company_handbook", "k": 3}, # 检索3个最相关的块 "generator": {"input": query} # 生成器的初始输入 }) print(f"生成的回答：{result['generator']}")

这个DagPipeline清晰地定义了RAG的完整逻辑。ToolRewriter节点可以优化用户查询（例如，将“年假政策”重写为“年度带薪休假规定”），提升检索质量。ChatPromptTemplate节点负责将检索到的文档块和用户问题组装成模型能理解的提示词。整个流程像一个可视化的工作流，每个模块职责单一，连接关系明确，无论是调试、优化还是扩展（比如在检索后加入一个重排序节点）都非常直观。

6. 实战避坑指南与高级技巧

经过几个项目的实际使用，我积累了一些经验和踩过的坑，在这里分享给你，希望能帮你少走弯路。

6.1 智能体与工具设计中的常见陷阱

陷阱一：工具描述不清导致模型误用@tool装饰器定义的函数，其文档字符串（docstring）和参数类型注解至关重要。模型依靠这些信息来决定是否以及如何调用工具。模糊的描述会导致模型不理解工具用途或传错参数。

反面例子：def fetch_data(query):参数和返回类型不明。

最佳实践：

@tool def search_product_inventory(product_id: str, warehouse: str = "default") -> Dict[str, Any]: """ 根据产品ID和仓库代码查询实时库存信息。 Args: product_id: 产品的唯一标识符，格式为'PROD-XXXXX'。 warehouse: 仓库代码，如 'WH-01'。默认为'default'主仓库。 Returns: 一个字典，包含产品ID、仓库、当前库存量、在途数量等信息。 如果未找到，返回空字典。 """ # ... 实现逻辑

清晰的描述能极大提升工具调用的准确率。

陷阱二：智能体陷入循环或无关对话当智能体拥有多个工具或能调用其他智能体时，有时它会陷入无意义的工具调用循环，或者开始与用户讨论“我该用哪个工具”而不是直接解决问题。

解决方案：精心设计system_prompt。明确指令其角色、目标和约束。

system_prompt = """你是一个高效的客户支持助手。你的首要目标是快速准确地解决用户问题。 1. 仔细分析用户问题，判断是否需要使用工具。 2. 如果需要，**直接调用最相关的工具**，不要询问用户确认。 3. 根据工具返回的结果，给出简洁、有帮助的最终答案。 4. 如果工具无法解决问题，请坦诚告知用户，并提供替代方案或建议联系人工客服。 绝对不要与用户讨论内部工具的使用流程。""" agent = Agent(name="support", client=client, tools=[tool1, tool2], system_prompt=system_prompt)

陷阱三：上下文过长导致性能下降和成本激增多轮对话或检索到大量文档块时，提示词会非常长，导致响应变慢、Token消耗剧增，甚至可能超过模型上下文窗口限制。

应对策略：
1. 使用有状态的记忆管理：Datapizza AI的Agent支持记忆（Memory）功能。确保你合理配置记忆的保留策略，例如只保留最近N轮对话，或自动总结历史对话。
2. 优化检索：在RAG中，不要盲目返回大量文档块（k值不宜过大）。可以尝试使用Reranker模块对初步检索结果进行重排序，只将最相关的少量片段送入提示词。
3. 选择适合长上下文的模型：对于需要处理超长文档的场景，考虑使用支持128K甚至更长上下文的模型。

6.2 生产环境部署的关键考量

1. 错误处理与重试机制网络波动、API限流、模型暂时不可用在生产中很常见。框架底层的客户端可能包含基础重试，但对于关键业务，你需要实现更健壮的策略。

建议：在调用agent.run或pipeline.run的外层，使用tenacity等重试库，并设置指数退避。

import tenacity from openai import RateLimitError, APIError @tenacity.retry( stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(multiplier=1, min=4, max=10), retry=tenacity.retry_if_exception_type((RateLimitError, APIError)), ) def robust_agent_query(agent, question): return agent.run(question) response = robust_agent_query(my_agent, "重要问题")

2. 监控与可观测性虽然框架内置了OpenTelemetry追踪，但在生产环境中，你需要将它集成到你的监控体系。

操作步骤：
1. 安装opentelemetry-sdk和导出器（如opentelemetry-exporter-jaeger）。
2. 在应用启动时配置TraceProvider，将数据导出到Jaeger、Datadog或你公司的监控平台。
3. 为关键业务操作（如订单处理、客户查询）设置唯一的追踪名称，便于在监控端筛选和报警。
4. 监控核心指标：每次调用的延迟、Token消耗、工具调用成功率。这些数据是优化成本和性能的基础。

3. 配置管理与安全永远不要将API密钥、数据库连接字符串等秘密信息硬编码。

标准做法：使用环境变量或专门的密钥管理服务。对于复杂的配置（如多个模型的API密钥、向量数据库地址），使用Pydantic Settings等库进行类型安全的集中管理。
```
# config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str qdrant_host: str = "localhost" qdrant_port: int = 6333 model_name: str = "gpt-4o-mini" class Config: env_file = ".env" settings = Settings()
```
然后在代码中通过settings.openai_api_key引用。

6.3 性能优化技巧

1. 并行化工具调用如果一个智能体需要调用多个彼此独立的工具（例如，同时查询天气和航班信息），串行调用会拖慢整体响应。虽然框架本身可能不直接提供并行工具调用，但你可以通过设计多个专门智能体，并由一个协调者异步调用的模式来模拟。

思路：使用asyncio库并发运行多个agent.run调用，然后汇总结果。这需要你更精细地控制工作流，但能显著降低延迟。

2. 嵌入与检索优化对于RAG应用，检索速度和质量是瓶颈。

使用更快的嵌入模型：如text-embedding-3-small在速度和效果间取得了很好平衡。对于纯英文场景，可以试试BAAI/bge-small-en等开源模型，通过FastEmbed集成，速度极快且本地部署。
索引优化：确保向量数据库的索引类型（如HNSW）参数配置合理。对于海量数据（>100万条），可能需要调整ef_construction和M参数来平衡构建速度、查询速度和精度。
引入重排序：初步用向量检索召回Top K个结果（比如K=20），再用一个轻量级、高精度的交叉编码器（Cross-Encoder）模型对它们进行重排序，只取Top N个（比如N=3）送入LLM。这能大幅提升答案相关性。Datapizza AI的Reranker模块（如CohereReranker）就是干这个的。

3. 缓存策略对于频繁出现的相同或相似查询（例如，常见问题FAQ），重复调用LLM是巨大的浪费。