1. 项目概述：用Python构建多Agent协作系统
去年我在开发一个智能客服系统时，第一次尝试使用OpenAI Agents SDK构建多Agent协作架构。当时需要同时处理用户咨询、工单分类和知识库检索三个任务，传统单体架构已经无法满足需求。通过将不同功能拆分为独立Agent并建立协作机制，最终系统响应速度提升了40%，准确率提高了25%。
OpenAI Agents SDK提供了一套完整的工具链，让我们能够用Python快速构建具备自主决策和协作能力的AI Agent系统。与传统的单任务模型不同，多Agent系统可以：
- 并行处理复杂任务流
- 通过消息传递实现动态协作
- 根据上下文灵活调整行为
- 实现分布式决策和负载均衡
2. 核心概念解析
2.1 Agent基础架构
一个标准的Agent通常包含以下组件：
class BasicAgent: def __init__(self): self.memory = [] # 对话历史记忆 self.tools = [] # 可用工具集 self.persona = {} # 角色设定 async def run(self, input): # 处理输入并生成响应 response = await self.llm_call(input) return self.post_process(response)2.2 多Agent协作模式
常见的协作模式包括：
| 模式类型 | 适用场景 | 实现复杂度 |
|---|---|---|
| 主从架构 | 任务分解 | ★★☆ |
| 对等网络 | 分布式决策 | ★★★ |
| 黑板模型 | 知识共享 | ★★★★ |
| 市场机制 | 资源分配 | ★★☆ |
3. 环境准备与SDK配置
3.1 Python环境要求
推荐使用Python 3.8+版本，关键依赖：
pip install openai==1.12.0 pip install asyncio pip install pydantic3.2 SDK初始化配置
from openai import OpenAI client = OpenAI( api_key="your_api_key", base_url="https://api.openai.com/v1", default_headers={"x-foo": "true"}, )注意：建议将API密钥存储在环境变量中，不要直接硬编码在代码里
4. 构建基础Agent
4.1 单Agent实现
class ResearchAgent: def __init__(self, role="researcher"): self.role = role self.tools = [web_search, read_pdf] async def execute(self, task): context = f"你是一个{self.role}，请完成以下任务：{task}" response = await client.chat.completions.create( model="gpt-4-turbo", messages=[{"role": "user", "content": context}], tools=self.tools ) return response.choices[0].message.content4.2 添加记忆功能
def add_memory_system(agent): memory = [] def decorator(func): async def wrapper(*args, **kwargs): memory.append({"input": args, "timestamp": time.time()}) result = await func(*args, **kwargs) memory[-1]["output"] = result return result return wrapper agent.execute = decorator(agent.execute) agent.memory = memory return agent5. 实现多Agent协作
5.1 消息路由系统
class MessageRouter: def __init__(self): self.agents = {} def register(self, name, agent): self.agents[name] = agent async def send(self, to, message): if to not in self.agents: raise ValueError(f"Unknown agent: {to}") return await self.agents[to].execute(message)5.2 协作工作流示例
async def research_workflow(question): # 初始化Agent researcher = ResearchAgent() analyst = AnalysisAgent() writer = WritingAgent() # 注册到路由 router = MessageRouter() router.register("research", researcher) router.register("analysis", analyst) router.register("writing", writer) # 执行协作流程 findings = await router.send("research", question) insights = await router.send("analysis", findings) return await router.send("writing", insights)6. 高级功能实现
6.1 动态工具调用
def tool_dispatcher(agent): async def dispatch(tool_name, *args): for tool in agent.tools: if tool.__name__ == tool_name: return await tool(*args) raise ValueError(f"Tool {tool_name} not found") return dispatch6.2 角色扮演配置
def configure_persona(agent, persona_config): agent.persona = { "style": persona_config.get("style", "professional"), "constraints": persona_config.get("constraints", []), "knowledge": persona_config.get("knowledge", {}) }7. 性能优化技巧
7.1 异步并发处理
async def parallel_execution(tasks): semaphore = asyncio.Semaphore(5) # 并发限制 async def limited_execute(task): async with semaphore: return await task.execute() return await asyncio.gather(*[limited_execute(t) for t in tasks])7.2 缓存策略实现
from functools import lru_cache @lru_cache(maxsize=1000) def cached_llm_call(prompt): return client.chat.completions.create( model="gpt-4-turbo", messages=[{"role": "user", "content": prompt}] )8. 调试与问题排查
8.1 常见错误处理
| 错误类型 | 解决方案 | 预防措施 |
|---|---|---|
| 速率限制 | 实现退避重试 | 监控token使用 |
| 工具调用失败 | 验证工具参数 | 添加类型检查 |
| 上下文溢出 | 分块处理 | 优化提示设计 |
8.2 日志记录系统
def setup_logging(agent): logger = logging.getLogger(agent.__class__.__name__) async def logged_execute(*args, **kwargs): logger.info(f"Input: {args}") try: result = await agent.execute(*args, **kwargs) logger.info(f"Output: {result}") return result except Exception as e: logger.error(f"Error: {str(e)}") raise agent.execute = logged_execute return agent9. 实际应用案例
9.1 客户支持系统
class SupportSystem: def __init__(self): self.router = MessageRouter() self.router.register("triage", TriageAgent()) self.router.register("technical", TechAgent()) self.router.register("billing", BillingAgent()) async def handle_request(self, user_input): category = await self.router.send("triage", user_input) return await self.router.send(category, user_input)9.2 数据分析流水线
async def data_analysis_pipeline(dataset): cleaners = [DataCleaner() for _ in range(3)] cleaned = await parallel_execution( [c.execute(dataset) for c in cleaners] ) analyzers = [Analyst() for _ in range(2)] results = await parallel_execution( [a.execute(cleaned) for a in analyzers] ) return await ReportGenerator().execute(results)10. 安全与监控
10.1 输入验证
def validate_input(func): async def wrapper(agent, input): if not isinstance(input, str): raise ValueError("Input must be string") if len(input) > 10000: raise ValueError("Input too long") return await func(agent, input) return wrapper10.2 监控仪表板
class Monitoring: def __init__(self): self.metrics = defaultdict(list) def track(self, metric, value): self.metrics[metric].append({ "value": value, "timestamp": time.time() }) def get_stats(self, metric): values = [m["value"] for m in self.metrics[metric]] return { "avg": sum(values)/len(values), "max": max(values), "min": min(values) }11. 部署与扩展
11.1 FastAPI集成
from fastapi import FastAPI app = FastAPI() agent_system = SupportSystem() @app.post("/query") async def handle_query(input: str): return await agent_system.handle_request(input)11.2 水平扩展策略
class AgentPool: def __init__(self, agent_class, size=5): self.pool = [agent_class() for _ in range(size)] self.counter = 0 async def execute(self, input): agent = self.pool[self.counter % len(self.pool)] self.counter += 1 return await agent.execute(input)12. 测试与验证
12.1 单元测试示例
@pytest.mark.asyncio async def test_research_agent(): agent = ResearchAgent() result = await agent.execute("Find latest AI papers") assert "paper" in result.lower()12.2 集成测试方案
@pytest.fixture def mock_agents(): router = MessageRouter() router.register("research", MockResearchAgent()) router.register("writing", MockWritingAgent()) return router @pytest.mark.asyncio async def test_full_workflow(mock_agents): result = await research_workflow("AI trends", router=mock_agents) assert len(result) > 10013. 性能基准测试
13.1 负载测试配置
async def load_test(agent, requests=100): tasks = [agent.execute(f"test {i}") for i in range(requests)] start = time.time() await asyncio.gather(*tasks) return time.time() - start13.2 优化前后对比
| 优化措施 | 平均响应时间 | 错误率 |
|---|---|---|
| 基础实现 | 1200ms | 5% |
| 添加缓存 | 800ms | 3% |
| 并行处理 | 400ms | 1% |
14. 最佳实践总结
- 角色定义清晰：每个Agent应该有明确的职责边界
- 消息格式统一：使用JSON等标准格式传递消息
- 错误隔离：单个Agent故障不应影响整个系统
- 监控全面：跟踪延迟、错误率和资源使用情况
- 版本控制：对Agent行为变更进行版本管理
15. 未来改进方向
在实际项目中，我发现以下几个方向值得深入探索：
- 实现Agent的自我优化能力
- 开发可视化编排工具
- 添加强化学习机制
- 优化Agent间通信协议
- 实现动态负载均衡
通过持续迭代，我们团队已经将Agent系统的任务处理能力提升了3倍。建议新手从简单的主从架构开始，逐步扩展到更复杂的协作模式。
