基于agents框架构建AI智能体：从单智能体问答到多智能体协作系统

1. 项目概述与核心价值

最近在探索AI应用落地的过程中，我花了大量时间研究各种开源框架，试图找到一个既能快速上手、又能支撑复杂业务逻辑的智能体开发方案。在这个过程中，一个名为alexleekt/agents的项目引起了我的注意。它不是一个简单的工具库，而是一个旨在构建“智能体操作系统”的框架。简单来说，它试图为AI智能体提供一个标准化的运行环境、一套清晰的开发范式以及一系列开箱即用的核心能力，让开发者能像搭积木一样，快速构建出具备自主感知、决策和执行能力的AI应用。

这个项目的核心价值在于“降本增效”。对于个人开发者或小团队而言，从零开始构建一个具备记忆、工具调用、多智能体协作等能力的系统，需要投入巨大的工程成本。agents框架将这些通用能力抽象成模块，并提供了一套统一的API，开发者只需关注业务逻辑本身。对于企业而言，它提供了一种将大语言模型（LLM）能力系统化、产品化的路径，使得构建客服助手、数据分析机器人、自动化流程引擎等应用变得更加标准化和可维护。它解决的不仅仅是“如何调用API”的问题，更是“如何设计一个可持续运行、可扩展、可观测的AI系统”的工程难题。

2. 核心架构与设计哲学拆解

2.1 分层架构：从内核到应用

深入分析agents的代码和设计文档，我发现其架构清晰地分为几个层次，这种分层设计保证了系统的灵活性和可扩展性。

内核层（Core）：这是框架的基石，定义了最核心的抽象。主要包括Agent（智能体）、Tool（工具）、Memory（记忆）等基础类。这里的Agent不是一个黑盒，而是一个由Role（角色定义）、Goal（目标）、LLM（大语言模型驱动）和一系列Tools构成的可配置实体。这种设计意味着，智能体的“性格”和“能力”是解耦的，你可以为一个“数据分析师”角色配置不同的工具集，或者让同一个工具被不同角色的智能体使用。

运行时层（Runtime）：这一层负责智能体的生命周期管理和任务执行。它包含了任务调度、事件循环、状态管理等功能。一个关键的设计是采用了异步（Async）优先的模式，这对于需要同时处理多个用户请求、或智能体需要等待外部API响应的I/O密集型场景至关重要。运行时层确保了智能体在执行复杂、多步骤任务时，能够高效地管理其内部状态和外部交互。

模块层（Modules）：这是框架“开箱即用”特性的体现。项目预置了许多实用的模块，例如：

• 记忆模块：包括短期的工作记忆（用于保持对话上下文）和长期的向量数据库记忆（用于知识存储和检索），这解决了LLM本身“健忘”和上下文长度有限的问题。
• 工具模块：内置了如网络搜索、文件读写、代码执行、API调用等常见工具。更重要的是，它提供了极其简便的工具定义和注册机制，让你用几行代码就能将任何一个函数“包装”成智能体可以调用的工具。
• 通信模块：用于实现多智能体之间的协作。智能体可以通过消息队列或直接的发布/订阅模式进行通信，这为构建分工明确的智能体团队（如一个负责检索，一个负责总结，一个负责格式化输出）奠定了基础。

应用层：在这一层，开发者利用下层提供的所有能力，组合出具体的业务应用。框架鼓励采用Agent作为应用的基本构建单元，整个应用可能就是一个智能体，也可能是一个由多个智能体组成的协作网络。

2.2 设计哲学：约定优于配置与组合性

agents框架深受现代软件工程思想的影响，其设计哲学可以概括为两点：

1. 约定优于配置（Convention Over Configuration）：框架为智能体定义了标准的运行流程（感知->规划->执行->反思），并提供了默认的实现。开发者不需要从头编写这些流程的代码，只需要通过配置（如YAML文件）或简单的代码来覆盖默认行为即可。这极大地减少了样板代码，让开发者能快速启动一个可工作的智能体。

2. 强大的组合性（Composability）：这是我认为该框架最精妙的地方。智能体、工具、记忆都是可插拔的组件。你可以像组装乐高一样，将一个具备网络搜索能力的工具赋予一个“市场调研”智能体，再将这个智能体与一个“报告撰写”智能体通过通信模块连接起来，形成一个自动化调研流水线。这种基于组件的设计，使得系统非常灵活，功能扩展和迭代变得非常容易。

注意：虽然框架提供了很多默认设置，但要充分发挥其威力，理解其底层抽象是必要的。不要仅仅满足于跑通示例，花时间研究Agent基类和Tool基类的源码，会让你在自定义功能时事半功倍。

3. 从零开始：构建你的第一个智能体

理论说了这么多，我们来点实际的。下面我将带你一步步使用agents框架，构建一个简单的“个人知识库问答助手”。这个智能体的目标是：回答用户关于特定知识领域（比如你个人的技术笔记）的问题，它需要能检索本地知识库，并结合联网搜索能力来补充最新信息。

3.1 环境搭建与初始化

首先，确保你的Python环境在3.8以上。创建一个新的虚拟环境是一个好习惯。

# 创建并激活虚拟环境（以conda为例） conda create -n ai-agents python=3.10 conda activate ai-agents # 安装 agents 框架 pip install agents-ai # 安装常用的额外依赖，如OpenAI、向量数据库客户端等 pip install openai chromadb

安装完成后，我们初始化一个项目。框架推荐一定的项目结构，但并非强制。一个清晰的结构有助于管理：

my_knowledge_agent/ ├── config/ │ └── agent_config.yaml # 智能体配置文件 ├── data/ │ └── my_notes.pdf # 你的知识文档 ├── tools/ │ └── custom_tools.py # 自定义工具 ├── main.py # 应用入口 └── requirements.txt

3.2 配置核心组件：模型、记忆与工具

接下来，我们通过代码和配置来定义智能体的核心部件。我们选择使用YAML进行配置，因为它更清晰、易于管理。

1. 模型配置 (config/agent_config.yaml)： 我们需要告诉框架使用哪个LLM。这里以OpenAI的GPT-4为例。

llm: provider: "openai" model: "gpt-4-turbo-preview" api_key: "${OPENAI_API_KEY}" # 建议从环境变量读取 temperature: 0.1 # 对于问答任务，低温度输出更稳定

2. 记忆系统实现： 记忆是智能体的核心。我们需要一个长期记忆来存储知识库。这里使用Chroma作为向量数据库。

在main.py中，我们初始化记忆：

import os from agents import Agent, Runner from agents.memory import VectorMemory import chromadb from chromadb.config import Settings # 初始化向量数据库客户端 chroma_client = chromadb.Client(Settings( chroma_db_impl="duckdb+parquet", persist_directory="./chroma_db" # 数据持久化目录 )) # 创建向量记忆实例 vector_memory = VectorMemory( client=chroma_client, collection_name="my_knowledge_base", embedding_model="text-embedding-3-small" # 使用的嵌入模型 ) # 假设我们有一个函数来处理文档并存入向量库 def init_knowledge_base(): from langchain.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader = PyPDFLoader("./data/my_notes.pdf") documents = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) chunks = text_splitter.split_documents(documents) # 将文本块添加到向量记忆 for chunk in chunks: vector_memory.add(chunk.page_content, metadata={"source": chunk.metadata.get("source", "unknown")}) print("知识库初始化完成。")

3. 工具定义与集成： 我们需要两个工具：一个用于检索本地知识库，一个用于联网搜索。

在tools/custom_tools.py中：

from agents import Tool import requests from typing import Optional class KnowledgeSearchTool(Tool): """用于检索本地知识库的工具""" name = "knowledge_search" description = "在本地知识库中搜索与问题相关的信息。当用户询问关于特定知识领域（如个人笔记、项目文档）的问题时使用此工具。" def __init__(self, vector_memory): super().__init__() self.memory = vector_memory async def run(self, query: str) -> str: """执行检索""" results = self.memory.search(query, top_k=3) # 返回最相关的3条 if not results: return "在知识库中未找到相关信息。" context = "\n---\n".join([res['content'] for res in results]) return f"从知识库中检索到以下相关信息：\n{context}" class WebSearchTool(Tool): """用于联网搜索的工具（示例，需接入真实搜索API如Serper、Tavily）""" name = "web_search" description = "在互联网上搜索最新信息。当问题涉及实时信息、新闻或知识库中找不到的内容时使用。" def __init__(self, api_key: Optional[str] = None): super().__init__() # 这里以Serper API为例，你需要注册并获取API Key self.api_key = api_key or os.getenv("SERPER_API_KEY") self.base_url = "https://google.serper.dev/search" async def run(self, query: str) -> str: headers = {"X-API-KEY": self.api_key, "Content-Type": "application/json"} payload = {"q": query, "gl": "us", "hl": "en"} try: response = requests.post(self.base_url, json=payload, headers=headers, timeout=10) response.raise_for_status() data = response.json() # 简化处理，提取前几条结果的摘要 snippets = [item.get('snippet', '') for item in data.get('organic', [])[:2]] return "网络搜索摘要：\n" + "\n".join([f"- {s}" for s in snippets if s]) except Exception as e: return f"网络搜索失败：{str(e)}"

3.3 智能体组装与任务执行

现在，我们将所有部件组装起来，创建智能体并运行它。

在main.py中继续：

# ... 之前的代码（记忆初始化等） from tools.custom_tools import KnowledgeSearchTool, WebSearchTool # 1. 初始化工具 knowledge_tool = KnowledgeSearchTool(vector_memory=vector_memory) web_search_tool = WebSearchTool() # 2. 创建智能体 my_agent = Agent( name="KnowledgeAssistant", role="你是一个专业的个人知识库助手，擅长根据本地知识库和网络信息回答技术问题。", goal="准确、全面地回答用户的问题，优先使用本地知识库，必要时补充最新网络信息。", tools=[knowledge_tool, web_search_tool], # 赋予智能体工具 memory=vector_memory, # 赋予长期记忆 # llm配置可以从yaml文件加载，这里简化处理 llm_config={ "provider": "openai", "model": "gpt-4-turbo-preview", "temperature": 0.1 } ) # 3. 运行智能体 async def main(): # 初始化知识库（首次运行需要） # init_knowledge_base() print("知识库助手已启动，请输入您的问题（输入'退出'或'quit'结束）：") while True: user_input = input("\n您: ") if user_input.lower() in ['退出', 'quit', 'exit']: print("助手：再见！") break # 使用Runner来运行智能体处理单个任务 result = await Runner.run(my_agent, user_input) print(f"\n助手：{result.final_output}") # 如果你想查看智能体的思考过程（使用了哪些工具，输出了什么） # for step in result.steps: # print(f"[步骤] {step.action} -> {step.observation}") if __name__ == "__main__": import asyncio asyncio.run(main())

这个简单的智能体已经具备了核心能力：它收到问题后，会先利用knowledge_search工具检索本地知识库，如果信息不足或需要最新动态，它会决定调用web_search工具，最后综合所有信息生成回答。agents框架自动管理了工具调用的流程、LLM的交互以及上下文的维护。

4. 进阶实战：构建多智能体协作系统

单一智能体能力有限，复杂的任务往往需要分工协作。agents框架对多智能体协作的支持是其一大亮点。让我们设计一个“内容创作团队”，由三个智能体组成：研究员、撰稿人、审校员。

4.1 定义智能体团队角色与工作流

我们的工作流是线性的：研究员收集信息 -> 撰稿人撰写初稿 -> 审校员润色修改。每个智能体都有明确的职责和不同的工具集。

from agents import Agent, Team from agents.communication import MessageQueue # 1. 创建通信通道 mq = MessageQueue() # 2. 定义并创建各个智能体 researcher = Agent( name="Researcher", role="信息研究员，擅长从网络和知识库中快速查找、筛选和总结信息。", goal="为撰稿人提供准确、全面、结构化的信息摘要。", tools=[web_search_tool, knowledge_tool], # 研究员需要搜索工具 llm_config={"model": "gpt-4-turbo-preview", "temperature": 0.2} ) writer = Agent( name="Writer", role="内容撰稿人，擅长将零散信息组织成结构清晰、语言流畅的文章。", goal="基于研究员提供的信息，撰写一篇符合要求的初稿。", tools=[], # 撰稿人可能不需要外部工具，专注于写作 llm_config={"model": "gpt-4-turbo-preview", "temperature": 0.7} # 温度稍高，创意更强 ) reviewer = Agent( name="Reviewer", role="内容审校员，擅长检查语法错误、优化措辞、确保风格一致。", goal="对撰稿人提供的初稿进行润色、修改和提升，输出最终稿。", tools=[], # 审校员也不需要外部工具 llm_config={"model": "gpt-4-turbo-preview", "temperature": 0.3} ) # 3. 创建团队，并定义工作流 content_team = Team( name="ContentCreationTeam", agents=[researcher, writer, reviewer], communication_bus=mq, workflow="sequential" # 顺序工作流：研究员 -> 撰稿人 -> 审校员 ) # 4. 为团队设置全局目标 async def create_content(topic: str): team_task = f“请围绕‘{topic}’这个主题，创作一篇800字左右的科普文章。” # 运行团队任务 final_result = await Runner.run(content_team, team_task) # 我们可以查看每个智能体的输出 print(f“最终文章：\n{final_result.final_output}\n”) print(“=== 工作流程记录 ===”) # 这里可以遍历团队的交互记录，查看协作过程

在这个设计中，Team对象负责协调。当收到任务后，它会根据workflow的定义，先将任务传递给Researcher。Researcher完成工作后，其输出会通过MessageQueue自动传递给Writer，以此类推。整个过程无需手动干预，框架管理了任务的路由和智能体间的通信。

4.2 实现自定义工作流与通信模式

顺序工作流只是最简单的一种。agents框架允许你定义更复杂的交互模式，例如基于事件的协作或竞争机制。

假设我们想让Researcher和Writer先并行工作（一个查资料，一个构思大纲），然后再由Reviewer整合。这需要更精细的控制。我们可以利用框架提供的Event和Condition机制，或者直接通过编程方式控制流程：

async def parallel_workflow(topic: str): """一个简化的并行-串行工作流示例""" # 步骤1：并行执行研究和大纲构思 research_task = f“请收集关于‘{topic}’的详细资料和最新动态。” outline_task = f“请为关于‘{topic}’的文章设计一个详细大纲。” # 使用 asyncio.gather 并行运行两个智能体 import asyncio research_future = Runner.run(researcher, research_task) outline_future = Runner.run(writer, outline_task) research_result, outline_result = await asyncio.gather(research_future, outline_future) # 步骤2：将并行结果交给撰稿人撰写全文 draft_task = f“请基于以下资料和大纲，撰写完整文章。\n资料：{research_result.final_output}\n大纲：{outline_result.final_output}” draft_result = await Runner.run(writer, draft_task) # 步骤3：审校 final_result = await Runner.run(reviewer, f“请审校并润色以下文章：\n{draft_result.final_output}”) return final_result

这种灵活性使得agents框架能够适应从简单的线性流程到复杂的动态决策图等各种业务场景。

5. 生产环境部署与性能优化

将实验性的智能体转化为稳定可靠的生产服务，需要考虑一系列工程问题。agents框架本身提供了一些支撑，但更多的需要开发者自己设计。

5.1 部署架构考量

对于轻量级应用，你可以使用像FastAPI这样的框架将智能体包装成HTTP API服务。

# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .agent_instance import my_agent # 导入你之前创建好的智能体实例 import asyncio app = FastAPI(title="Knowledge Agent API") class QueryRequest(BaseModel): question: str session_id: str = None # 用于区分不同对话会话 @app.post(“/ask”) async def ask_question(request: QueryRequest): try: # 这里可以加入会话管理，将session_id与特定的记忆实例关联 result = await Runner.run(my_agent, request.question) return {“answer”: result.final_output, “session_id”: request.session_id} except Exception as e: raise HTTPException(status_code=500, detail=f“Agent execution failed: {str(e)}”)

对于高并发场景，你需要考虑：

• 智能体实例池：避免为每个请求创建新的智能体对象，开销巨大。可以维护一个智能体实例池。
• 异步与并发：确保你的所有工具（尤其是网络请求）和智能体逻辑都是异步的，以避免阻塞事件循环。
• 状态外部化：将智能体的记忆（对话历史、知识库）完全存储到外部数据库（如Redis、PostgreSQL），使服务本身成为无状态的，便于水平扩展。

5.2 监控、评估与持续改进

一个AI应用上线后，监控其表现至关重要。

1. 日志记录：详细记录每个用户查询、智能体的思考过程（Chain-of-Thought）、工具调用详情、最终输出、耗时和Token使用量。这不仅是排查问题的依据，也是优化和评估的数据基础。
2. 评估体系：
   • 人工评估：定期抽样检查，评估回答的准确性、相关性和有用性。
   • 自动评估：设计一些启发式规则或利用另一个LLM作为裁判，对输出进行初步筛选（例如，检查是否包含“我不知道”之类的逃避回答，或检测明显的事实错误）。
   • 业务指标：如果智能体用于客服，可以关联解决率、用户满意度评分；如果用于销售，可以关联转化率。
3. 反馈闭环：建立机制让用户对回答进行点赞/点踩。这些反馈数据可以用于后续的模型微调（RAG中的检索器优化、重排序模型训练）或提示词（Prompt）的迭代。

5.3 成本与性能优化技巧

LLM API调用是主要成本。以下是一些优化策略：

• 缓存层：对频繁出现的、答案固定的问题（如“你们的营业时间？”），在智能体调用LLM之前，先查询缓存。可以使用Redis或Memcached。
• 小模型优先策略：设计一个路由层。简单问题（如问候、简单查询）使用便宜快速的模型（如gpt-3.5-turbo），复杂问题再路由到GPT-4。
• 优化提示词（Prompt）：清晰、具体的提示词能减少LLM的“胡思乱想”，提高输出质量，有时还能缩短响应长度，节省Token。持续进行A/B测试优化你的提示词模板。
• 向量检索优化：
  • 分块策略：文本分块的大小和重叠度对检索质量影响巨大。需要根据你的文档类型（技术文档、对话记录、新闻）进行实验调整。
  • 重排序（Re-ranking）：初步检索出10个相关块后，使用一个更小、更快的重排序模型（如BGE-reranker）对这10个块进行精排，只将前3个最相关的块送入LLM上下文，这能在不损失精度的情况下显著降低Token消耗。

实操心得：在预生产环境中，一定要对智能体进行“压力测试”。模拟大量并发请求，观察内存泄漏、响应时间衰减和API错误率。我曾遇到过因为工具函数中一个同步的网络请求未做超时处理，导致整个事件循环被阻塞的情况。务必确保所有I/O操作都是异步且具有鲁棒性的。

6. 常见问题排查与调试技巧

在实际开发中，你肯定会遇到各种问题。下面是一些常见坑点和排查思路。

6.1 智能体“发呆”或不调用工具

现象：智能体直接基于自身知识生成回答，而不去调用你给它配置的工具。排查步骤：

1. 检查工具描述：LLM决定是否调用工具，很大程度上依赖于你为工具定义的description字段。确保描述清晰、具体，说明了工具的用途和适用场景。例如，“搜索网络信息”就比“搜索工具”要好得多。
2. 启用详细日志：在运行智能体时，设置更高的日志级别，或像之前示例一样打印出result.steps。查看智能体的思考链，看它是否在决策步骤中考虑了工具但最终放弃了，还是完全没考虑。
3. 调整提示词：在智能体的role或goal描述中，明确强调“你必须使用提供的工具来获取信息”或“在回答关于X的问题前，请先使用Y工具进行查询”。
4. 测试工具本身：单独测试你的工具函数，确保其输入输出符合预期，没有抛出异常。一个失败的工具调用可能会让LLM在后续对话中避免使用它。

6.2 记忆检索效果不佳

现象：智能体检索到的知识库内容与问题不相关，导致回答质量差。排查步骤：

1. 检查嵌入模型：不同的嵌入模型（如text-embedding-ada-002vstext-embedding-3-small）在不同语种和领域的表现差异很大。确保你使用的模型适合你的文本类型。
2. 审视文本分块：这是影响检索效果最关键的因素之一。
   • 块太大：包含太多信息，噪声多，精度低。
   • 块太小：语义不完整，召回率低。
   • 建议：尝试不同的分块大小（如256, 512, 1024 tokens）和重叠度（如10%-20%）。对于结构化文档（如Markdown），可以尝试按标题进行分块。
3. 尝试混合检索：除了向量检索，可以加入关键词（如BM25）检索，并将两者结果融合（Hybrid Search）。Chroma等数据库支持这种模式。
4. 添加元数据过滤：在存储文档时，为其添加元数据（如“章节名”、“文档类型”、“创建日期”）。在检索时，可以结合元数据过滤，缩小搜索范围。

6.3 多智能体协作卡死或循环

现象：智能体之间互相传递消息，但无法达成共识或完成任务，陷入死循环。排查步骤：

1. 明确停止条件：为团队或工作流设置明确的任务完成条件。例如，在团队目标中写明“当生成一篇经过审校的最终文章后，任务完成”。
2. 引入监督者（Supervisor）：创建一个更高级别的“经理”智能体，负责协调团队、评估进度，并在必要时中断循环或做出仲裁。
3. 限制交互轮次：在代码层面设置一个最大的消息传递轮次限制，超过轮次则强制终止，并返回当前最佳结果或错误信息。
4. 分析通信日志：详细记录每个智能体的输入和输出。循环通常是因为某个智能体的输出格式不符合下一个智能体的预期，或者任务目标本身模糊不清。

6.4 API限速与错误处理

现象：服务因LLM API调用频繁被限速或因网络问题失败。解决方案：

1. 实现重试与退避：对所有外部API调用（LLM、工具）包裹具有指数退避（Exponential Backoff）机制的重试逻辑。可以使用tenacity或backoff库。
2. 设置合理的超时：为每个LLM调用和工具调用设置明确的超时时间，避免一个慢请求拖垮整个系统。
3. 使用队列：对于非实时性要求极高的场景，可以将用户请求放入队列（如RabbitMQ,Celery），由后台工作进程按可控的速率消费，平滑请求峰值。
4. 熔断与降级：当某个上游服务（如特定的LLM提供商）持续失败时，可以“熔断”该路由，暂时将流量切换到备用模型或返回一个友好的降级提示（如“系统正在优化，请稍后再试”）。

开发基于agents这类框架的应用，是一个典型的“三分调模型，七分调系统”的过程。框架解决了智能体范式和基础架构的问题，但构建一个稳定、高效、可控的AI应用，仍然需要扎实的软件工程能力和对业务场景的深刻理解。从简单的单智能体问答开始，逐步迭代到复杂的多智能体工作流，并在过程中持续监控、评估和优化，是通往成功的务实路径。