你是否曾想过让一个AI助手帮你写代码、查资料、分析数据但它却常常“卡壳”要么理解不了复杂需求要么输出的结果离题万里这背后往往不是模型能力不行而是任务拆解和执行的逻辑出了问题。单个AI智能体Agent就像一个全能的实习生什么都会一点但面对一个需要多步骤、多领域协作的复杂项目时它很容易陷入混乱。今天要介绍的Oh My Subagents正是为了解决这个痛点而生。它不是一个全新的AI模型而是一个精巧的多智能体协作框架。它的核心思想是“分而治之”将一个复杂的顶层任务自动分解成一系列由不同专业子智能体Subagent执行的子任务并通过一个中央协调器Orchestrator来管理和汇总结果。简单来说它让AI从“单打独斗”变成了“团队作战”。你只需要告诉它最终目标比如“开发一个带用户登录的待办事项Web应用”它就能自动规划出“设计数据库”、“编写后端API”、“实现前端页面”、“处理用户认证”等子任务并分派给对应的子智能体去执行最后将成果整合起来。本文将带你深入探索 Oh My Subagents。我们不仅会厘清多智能体系统的核心概念更会通过一个从零开始的完整实战项目手把手教你搭建环境、编写配置、运行智能体并剖析其背后的工作流与通信机制。更重要的是我们会揭示在实际使用中可能遇到的“坑”比如任务循环、幻觉问题、成本控制并给出经过验证的最佳实践。无论你是想探索AI应用开发新范式的开发者还是正在寻找提升自动化效率工具的技术负责人这篇文章都将提供可直接落地的解决方案。1. 这篇文章真正要解决的问题在AI应用开发中我们经常面临一个困境大语言模型LLM在单轮对话或简单指令跟随上表现出色但一旦任务变得复杂、多步骤、需要长期记忆或领域专业知识时其表现就大打折扣。开发者不得不手动编写大量的提示词Prompt工程代码预先定义好所有可能的执行路径这种“硬编码”的方式不仅繁琐而且缺乏灵活性。Oh My Subagents 瞄准的正是“复杂任务自动化”这一核心难题。它试图通过一套标准化的框架将任务分解、子智能体调度、上下文传递和结果整合这些重复性工作抽象出来。开发者不再需要关注“如何让AI一步步思考”而是专注于定义“顶级目标是什么”以及“有哪些可用的专业工具子智能体”。对于读者而言阅读本文将解决以下几个具体问题认知门槛理解多智能体系统Multi-Agent System与传统单智能体或简单函数调用的本质区别。实操空白市面上概念讲解多但完整、可运行的代码示例少。本文将提供一个端到端的可复现案例。工程化盲区多智能体系统在真实项目中如何管理状态如何保证任务不陷入死循环如何控制API调用成本这些工程细节往往是失败的关键。技术选型困惑与 LangChain、AutoGPT 等项目相比Oh My Subagents 的定位和优势是什么它更适合什么场景本文的目标读者是有一定Python基础对AI应用开发感兴趣并希望将自动化能力提升到新层次的开发者。接下来的内容我们将从原理到实践彻底拆解这个框架。2. 基础概念与核心原理在深入代码之前我们必须建立清晰的概念模型。Oh My Subagents 架构中的几个核心角色理解了它们就理解了整个系统的工作流。2.1 核心角色定义角色职责类比Orchestrator (协调器)系统的“大脑”。接收用户初始请求将其分解为子任务决定执行顺序分派给合适的 Subagent并汇总最终结果。项目经理Subagent (子智能体)系统的“四肢”。专门负责某一类具体任务如写Python代码、查询数据库、分析数据。每个Subagent都有自己的系统提示词System Prompt和可用工具。专业工程师前端、后端、DBATool (工具)Subagent 可以调用的具体函数。例如“执行Python代码”、“运行SQL查询”、“调用外部API”。这扩展了AI的能力边界。工程师手中的IDE、命令行或软件Working Memory (工作记忆)一个共享的上下文存储。Orchestrator 和 Subagents 都可以从中读取信息或写入新的发现、代码、数据结果。这是智能体间通信的桥梁。团队的共享文档或协作白板Task Queue (任务队列)由Orchestrator维护的一个待处理子任务列表。系统会循环地从队列中取出任务执行直到队列为空或达到停止条件。项目看板Kanban2.2 工作流程详解整个系统的工作流程是一个典型的“规划-执行-评估”循环可以概括为以下几步任务输入用户向Orchestrator提出一个复杂请求例如“分析CSV文件sales.csv找出销量最好的产品并用图表展示”。任务规划Orchestrator 分析该请求利用其内置的LLM能力将之分解为一系列有序的子任务并放入任务队列。示例分解[“加载并查看sales.csv文件结构” “计算每个产品的总销量” “找出销量最高的产品” “生成销量排名的柱状图”]任务执行Orchestrator 从队列中取出第一个子任务根据任务类型如“数据加载”、“计算”、“可视化”选择最匹配的Subagent如“数据分析Agent”、“代码执行Agent”。工具调用被选中的Subagent 理解子任务决定需要调用哪个Tool如“pandas.read_csv”、“matplotlib.pyplot.bar”来完成工作。调用可能成功也可能失败。结果写入与循环Subagent 将执行结果成功的数据或失败的异常信息写入Working Memory。Orchestrator 检查当前子任务是否完成然后决定是继续执行队列中的下一个任务还是基于新发现的信息创建新的子任务例如发现数据有缺失新增“处理缺失值”任务。最终汇总当所有子任务完成且没有新任务产生时Orchestrator 从 Working Memory 中提取所有关键结果整合成一份完整的、面向用户的答复。这个流程的关键在于“动态规划”。与传统脚本不同Orchestrator 可以根据执行中间结果实时调整计划这使得系统能处理一些未预见的状况。2.3 与相关技术的对比为了更精准地定位 Oh My Subagents我们将其与几个常见技术进行对比技术/框架核心模式与 Oh My Subagents 的差异LangChain提供构建LLM应用所需的丰富“组件”Models, Prompts, Chains, Agents, Tools。更像一个“工具箱”。Oh My Subagents 是在 LangChain 等底层库之上封装好的一套特定范式多智能体协作的上层框架。它用 LangChain 来实现单个Agent的能力。AutoGPT一个追求完全自治的AI应用目标是给定一个目标AI能自我循环直到完成。AutoGPT 是 Oh My Subagents 理念的一个具体实现案例但通常更重、更复杂。Oh My Subagents 作为一个框架更轻量更关注提供可定制、可观察的协作机制。简单函数调用将LLM输出解析为结构化参数调用预定函数。函数调用是单次、被动的。Oh My Subagents 是主动、持续、多轮次的协作系统具备任务生成和调度能力。简单理解如果你用 LangChain 是自己在组装一台电脑那么用 Oh My Subagents 就是直接拿到了一台设计好用于“团队协作”的专用服务器。3. 环境准备与前置条件理论清晰后我们开始动手搭建。Oh My Subagents 是一个 Python 项目因此你需要一个可用的 Python 环境。3.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。本文演示基于 macOS/Linux 命令行Windows 用户建议使用 WSL2 或 Git Bash 以获得最佳体验。Python 版本Python 3.8 至 3.11。强烈推荐使用 Python 3.10 或 3.11以获得最佳的库兼容性。Python 3.12 可能因某些依赖库尚未适配而存在风险。包管理工具pip(现代 Python 已内置)。建议使用虚拟环境。API 密钥你需要一个OpenAI API 密钥或兼容 OpenAI API 的其他大模型服务如 Azure OpenAI, 国内的一些合规大模型平台的密钥。Oh My Subagents 的核心协调与执行能力依赖于大语言模型。3.2 创建虚拟环境与安装使用虚拟环境可以隔离项目依赖避免污染系统Python环境。# 1. 克隆 Oh My Subagents 仓库 (假设项目已开源在 GitHub) git clone https://github.com/your-username/oh-my-subagents.git # 请替换为实际仓库地址 cd oh-my-subagents # 2. 创建并激活虚拟环境 (以 venv 为例) python -m venv venv # 在 macOS/Linux 上激活 source venv/bin/activate # 在 Windows (CMD) 上激活 # venv\Scripts\activate # 3. 升级 pip 并安装项目依赖 pip install --upgrade pip # 安装核心依赖。如果项目有 requirements.txt则使用 # pip install -r requirements.txt # 此处我们假设需要安装以下典型依赖 pip install openai langchain langchain-openai sqlalchemy pandas matplotlib # 注实际依赖请以项目官方文档或 requirements.txt 为准3.3 配置 API 密钥安全地管理你的API密钥不要将其硬编码在代码中。推荐使用环境变量。# 在 macOS/Linux 的终端中设置环境变量 (临时仅当前会话有效) export OPENAI_API_KEYsk-your-actual-openai-api-key-here # 在 Windows (CMD) 中设置环境变量 # set OPENAI_API_KEYsk-your-actual-openai-api-key-here # 更持久的方法将上述命令添加到你的 shell 配置文件 (~/.bashrc, ~/.zshrc, 或 ~/.bash_profile) # echo export OPENAI_API_KEYsk-your-actual-key ~/.zshrc # source ~/.zshrc在你的Python代码中可以通过os.environ来读取这个密钥。import os api_key os.environ.get(OPENAI_API_KEY) if not api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量)4. 核心流程拆解构建你的第一个智能体团队现在我们开始构建一个具体的多智能体系统。我们的目标是创建一个能自动进行数据分析和可视化的团队。团队由以下角色构成Orchestrator总指挥理解用户需求并分解任务。Data Analyst Subagent数据分析专家擅长用Pandas处理数据。Code Executor Subagent代码执行专家负责安全地运行Python代码并返回结果。Visualization Subagent可视化专家擅长用Matplotlib生成图表。4.1 步骤一定义工具Tools工具是智能体能力的延伸。我们首先定义几个基础工具。# file: my_tools.py import pandas as pd import matplotlib.pyplot as plt import io import sys from contextlib import redirect_stdout, redirect_stderr from typing import Optional, Dict, Any def read_csv_file(file_path: str) - Optional[pd.DataFrame]: 读取CSV文件并返回DataFrame。 try: df pd.read_csv(file_path) return df except Exception as e: return f读取文件失败: {e} def describe_dataframe(df: pd.DataFrame) - Dict[str, Any]: 返回DataFrame的基本描述信息。 if df is None: return {error: DataFrame 为空} info { shape: df.shape, columns: df.columns.tolist(), dtypes: df.dtypes.astype(str).to_dict(), head: df.head().to_dict(orientrecords) } return info def execute_python_code(code_snippet: str) - str: 在一个受限的安全环境中执行一段Python代码。 返回标准输出和标准错误。 output_buffer io.StringIO() error_buffer io.StringIO() try: # 重定向输出和错误 with redirect_stdout(output_buffer), redirect_stderr(error_buffer): # 限制可用的全局变量增加安全性 safe_globals { __builtins__: __builtins__, pd: pd, plt: plt, df: None # 将在执行前注入 } exec(code_snippet, safe_globals) stdout output_buffer.getvalue() stderr error_buffer.getvalue() result f执行成功。\n标准输出:\n{stdout}\n标准错误:\n{stderr} if stdout or stderr else 执行成功无输出。 except Exception as e: result f执行出错: {e} return result def create_bar_chart(data_dict: dict, title: str, xlabel: str, ylabel: str) - str: 根据提供的数据字典生成一个柱状图并保存为图片。 try: names list(data_dict.keys()) values list(data_dict.values()) plt.figure(figsize(10, 6)) plt.bar(names, values) plt.title(title) plt.xlabel(xlabel) plt.ylabel(ylabel) plt.xticks(rotation45) plt.tight_layout() save_path output_chart.png plt.savefig(save_path) plt.close() return f图表已生成并保存至: {save_path} except Exception as e: return f生成图表失败: {e}关键点execute_python_code函数使用了exec这在生产环境中存在安全风险。这里仅作演示真实场景下必须使用沙箱环境如 Docker 容器或更严格的限制。工具函数应尽量保持纯净输入输出明确便于智能体理解和调用。4.2 步骤二创建子智能体Subagents每个子智能体本质上是一个具备特定系统提示词System Prompt和工具集的 LangChain Agent。# file: my_agents.py from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate import os from my_tools import read_csv_file, describe_dataframe, execute_python_code, create_bar_chart # 初始化LLM llm ChatOpenAI( modelgpt-3.5-turbo, # 或 gpt-4根据你的API权限选择 temperature0, # 降低随机性使输出更稳定 openai_api_keyos.environ.get(OPENAI_API_KEY) ) # 1. 创建数据分析子智能体 data_analysis_tools [ Tool( nameReadCSV, funcread_csv_file, description读取一个CSV文件并返回其内容。输入应为文件的路径字符串。 ), Tool( nameDescribeData, funcdescribe_dataframe, description描述一个Pandas DataFrame的基本信息包括形状、列名、数据类型和前几行数据。输入应是一个DataFrame对象。 ) ] data_analyst_prompt PromptTemplate.from_template( 你是一个专业的数据分析师。你的任务是帮助用户理解和处理数据。 你可以使用工具来读取CSV文件和查看数据集的描述信息。 当用户询问数据相关问题时请先使用工具获取数据信息再基于信息进行回答。 保持回答专业、简洁。 当前对话上下文 {context} 问题{input} 开始思考 ) data_analyst_agent create_react_agent(llmllm, toolsdata_analysis_tools, promptdata_analyst_prompt) data_analyst_executor AgentExecutor(agentdata_analyst_agent, toolsdata_analysis_tools, verboseTrue) # 2. 创建代码执行子智能体 code_execution_tools [ Tool( nameExecutePython, funcexecute_python_code, description执行一段Python代码并返回结果。输入应是一段有效的Python代码字符串。注意代码将在当前环境执行请确保安全。 ) ] code_executor_prompt PromptTemplate.from_template( 你是一个Python代码执行专家。你的唯一任务是安全地执行用户提供的Python代码片段并返回执行结果包括输出和错误。 你不应该修改代码或解释代码除非用户明确要求。 如果代码需要特定的变量如df请确保在执行前用户或上下文已提供。 当前对话上下文 {context} 代码片段{input} 开始执行 ) code_executor_agent create_react_agent(llmllm, toolscode_execution_tools, promptcode_executor_prompt) code_executor_executor AgentExecutor(agentcode_executor_agent, toolscode_execution_tools, verboseTrue) # 3. 创建可视化子智能体 viz_tools [ Tool( nameCreateBarChart, funccreate_bar_chart, description根据提供的数据字典生成柱状图。输入应为四个参数data_dict (字典), title (字符串), xlabel (字符串), ylabel (字符串)。 ) ] viz_agent_prompt PromptTemplate.from_template( 你是一个数据可视化专家。你的任务是根据用户提供的数据和需求生成合适的图表。 目前你擅长生成柱状图。请确保用户提供了生成图表所需的所有数据通常是键值对字典和标签信息。 当前对话上下文 {context} 用户需求{input} 开始工作 ) viz_agent create_react_agent(llmllm, toolsviz_tools, promptviz_agent_prompt) viz_executor AgentExecutor(agentviz_agent, toolsviz_tools, verboseTrue)关键点我们使用 LangChain 的create_react_agent来构建基于 ReAct 推理框架的智能体。AgentExecutor是实际运行智能体的对象verboseTrue会打印详细的思考过程便于调试。每个智能体都有高度定制化的系统提示词这决定了它的“专业领域”和行为模式。4.3 步骤三实现协调器Orchestrator与工作流协调器是系统的核心。它需要具备任务分解、智能体路由和上下文管理的能力。这里我们实现一个简化版本。# file: my_orchestrator.py from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from typing import List, Dict, Any import json class SimpleOrchestrator: def __init__(self, agents: Dict[str, Any], llm): 初始化协调器。 :param agents: 一个字典键为智能体角色名值为对应的AgentExecutor实例。 :param llm: 用于任务规划和路由的LLM。 self.agents agents self.llm llm self.working_memory {} # 简化的共享内存用字典实现 self.task_queue [] # 任务队列 def _plan_tasks(self, user_input: str) - List[str]: 利用LLM将用户输入分解为子任务列表。 planning_prompt f 你是一个高级任务规划师。请将用户的复杂请求分解成一系列清晰的、可顺序执行的子任务。 每个子任务应该能被以下专业角色之一处理 {, .join(self.agents.keys())} 用户请求{user_input} 请以JSON列表的形式输出子任务例如[任务1描述, 任务2描述, ...] 只输出JSON不要有其他文字。 messages [ SystemMessage(content你是一个高效的任务分解专家。), HumanMessage(contentplanning_prompt) ] response self.llm.invoke(messages) try: tasks json.loads(response.content) return tasks if isinstance(tasks, list) else [] except json.JSONDecodeError: # 如果LLM输出不是标准JSON尝试简单分割 return [task.strip() for task in response.content.split(\n) if task.strip()] def _route_agent(self, task: str) - str: 根据任务描述决定由哪个智能体处理。 routing_prompt f 请判断以下任务最适合由哪个专业智能体处理。可选的智能体有 {json.dumps(list(self.agents.keys()), indent2)} 任务描述{task} 请只返回智能体的名称不要有任何其他解释。 messages [ SystemMessage(content你是一个智能体路由专家。), HumanMessage(contentrouting_prompt) ] response self.llm.invoke(messages) agent_name response.content.strip() return agent_name if agent_name in self.agents else list(self.agents.keys())[0] # 默认返回第一个 def run(self, user_input: str) - Dict[str, Any]: 执行主工作流。 print(f[Orchestrator] 收到用户请求: {user_input}) # 1. 规划任务 self.task_queue self._plan_tasks(user_input) print(f[Orchestrator] 生成任务队列: {self.task_queue}) final_result {original_request: user_input, subtask_results: []} # 2. 循环执行任务队列 while self.task_queue: current_task self.task_queue.pop(0) print(f\n[Orchestrator] 处理任务: {current_task}) # 2.1 路由 assigned_agent_name self._route_agent(current_task) print(f[Orchestrator] 路由给智能体: {assigned_agent_name}) # 2.2 执行 # 构建上下文信息传递给智能体 context f工作记忆: {json.dumps(self.working_memory, indent2)}\n当前任务: {current_task} try: agent_executor self.agents[assigned_agent_name] # 注意这里简化了调用实际需要根据智能体类型调整输入格式 result agent_executor.invoke({input: current_task, context: context}) output result.get(output, No output) except Exception as e: output f智能体执行出错: {e} print(f[智能体 {assigned_agent_name}] 输出: {output[:200]}...) # 截断长输出 # 2.3 记录结果并更新工作记忆 task_result { task: current_task, agent: assigned_agent_name, output: output } final_result[subtask_results].append(task_result) # 简单地将最新结果存入记忆实际应更智能地提取关键信息 self.working_memory[ftask_{len(final_result[subtask_results])}] { description: current_task, result_summary: str(output)[:500] # 存储摘要 } # 2.4 可选基于结果动态添加新任务 # 这里可以加入逻辑例如如果输出包含“数据缺失”则添加“处理缺失值”任务 # if missing in output.lower(): # new_task 处理数据中的缺失值 # self.task_queue.append(new_task) # print(f[Orchestrator] 基于结果添加新任务: {new_task}) # 3. 生成最终汇总 final_result[working_memory_final] self.working_memory print(f\n[Orchestrator] 所有任务执行完毕。) return final_result关键点这个SimpleOrchestrator是一个极简实现真实框架如 Oh My Subagents的协调器会更复杂包含更优的任务分解算法、错误处理、循环检测等。_plan_tasks和_route_agent都依赖LLM这会产生额外的API调用和延迟但灵活性极高。working_memory目前是简单的字典生产系统可能需要向量数据库或更结构化的存储来管理长期、大量的上下文。5. 完整示例运行你的多智能体系统让我们将以上所有部分组合起来并模拟一个完整的用户请求。5.1 准备数据与主程序首先创建一个示例CSV文件sales_data.csv。# file: sales_data.csv Product,Month,Sales Laptop,Jan,120 Laptop,Feb,150 Laptop,Mar,180 Mouse,Jan,300 Mouse,Feb,320 Mouse,Mar,310 Keyboard,Jan,200 Keyboard,Feb,210 Keyboard,Mar,190 Monitor,Jan,80 Monitor,Feb,90 Monitor,Mar,95然后编写主程序来启动整个系统。# file: main.py import os from my_agents import data_analyst_executor, code_executor_executor, viz_executor, llm from my_orchestrator import SimpleOrchestrator def main(): # 1. 组装智能体团队 agents { DataAnalyst: data_analyst_executor, CodeExecutor: code_executor_executor, VisualizationExpert: viz_executor, } # 2. 初始化协调器 orchestrator SimpleOrchestrator(agentsagents, llmllm) # 3. 模拟用户请求 user_request 请分析当前目录下的 sales_data.csv 文件。 我需要知道 1. 数据的基本情况有哪些列多少行。 2. 每个产品Product的总销售额Sales是多少。 3. 哪个产品的总销售额最高。 4. 为每个产品的总销售额生成一个柱状图并保存为 product_sales.png。 print(*50) print(启动多智能体协作系统...) print(*50) # 4. 运行协调器 final_report orchestrator.run(user_request) # 5. 打印最终报告摘要 print(\n *50) print(任务执行最终报告) print(*50) print(f原始请求: {final_report[original_request][:100]}...) print(f\n共执行了 {len(final_report[subtask_results])} 个子任务:) for i, task_res in enumerate(final_report[subtask_results], 1): print(f {i}. [{task_res[agent]}] {task_res[task]}) print(f 输出摘要: {task_res[output][:150]}...) # 6. 检查生成的图表文件 if os.path.exists(product_sales.png): print(f\n✅ 图表文件 product_sales.png 已生成。) elif os.path.exists(output_chart.png): print(f\n 图表文件 output_chart.png 已生成默认名称。) else: print(f\n❌ 未找到生成的图表文件请检查可视化步骤的输出。) if __name__ __main__: # 确保设置了API密钥 if not os.environ.get(OPENAI_API_KEY): print(错误请设置 OPENAI_API_KEY 环境变量。) exit(1) main()5.2 运行与观察在终端中运行你的主程序# 确保在虚拟环境中且当前目录包含所有.py文件和sales_data.csv python main.py你将看到类似以下的输出具体内容因模型随机性略有不同 启动多智能体协作系统... [Orchestrator] 收到用户请求: 请分析当前目录下的 sales_data.csv 文件... [Orchestrator] 生成任务队列: [读取并查看 sales_data.csv 文件结构, 计算每个产品的总销售额, 找出总销售额最高的产品, 生成产品总销售额的柱状图] [Orchestrator] 处理任务: 读取并查看 sales_data.csv 文件结构 [Orchestrator] 路由给智能体: DataAnalyst Entering new AgentExecutor chain... 思考我需要先读取文件。我将使用ReadCSV工具。 动作ReadCSV 动作输入sales_data.csv 观察返回了一个DataFrame对象此处为简化实际为字典表示。 思考现在我需要描述这个DataFrame。使用DescribeData工具。 动作DescribeData 动作输入上一步得到的DataFrame 观察{shape: (12, 3), columns: [Product, Month, Sales], ...} 思考我有足够的信息回答。数据有12行3列列名是Product, Month, Sales。 动作最终答案 Finished chain. [智能体 DataAnalyst] 输出: 文件sales_data.csv已成功读取。数据集包含12行和3列。列名分别为Product产品、Month月份、Sales销售额。前几行数据显示了不同产品在各个月的销售情况。... ... (后续任务执行过程类似) ... [Orchestrator] 所有任务执行完毕。 任务执行最终报告 原始请求: 请分析当前目录下的 sales_data.csv 文件... 共执行了 4 个子任务: 1. [DataAnalyst] 读取并查看 sales_data.csv 文件结构 输出摘要: 文件sales_data.csv已成功读取。数据集包含12行和3列。列名分别为Product产品、Month月份、Sales销售额... 2. [CodeExecutor] 计算每个产品的总销售额 输出摘要: 执行成功。标准输出: 产品总销售额: Laptop 450, Mouse 930, Keyboard 600, Monitor 265... 3. [DataAnalyst] 找出总销售额最高的产品 输出摘要: 根据计算出的总销售额数据总销售额最高的产品是Mouse总销售额为930... 4. [VisualizationExpert] 生成产品总销售额的柱状图 输出摘要: 图表已生成并保存至: product_sales.png... ✅ 图表文件 product_sales.png 已生成。6. 运行结果与效果验证运行成功后你应该在项目目录下看到新生成的product_sales.png图表文件。用图片查看器打开应该能看到一个清晰的柱状图展示了 Laptop, Mouse, Keyboard, Monitor 四个产品的总销售额对比。如何验证系统工作正常流程完整性检查控制台输出确认4个子任务都被识别、路由并执行。结果正确性手动计算sales_data.csv中每个产品的 Sales 总和与智能体输出的结果Mouse: 930核对。打开生成的 PNG 文件确认图表标题、坐标轴标签和数据与预期一致。工作记忆在final_report变量中查看working_memory_final字段确认每个任务的结果摘要都被正确存储。错误处理你可以尝试修改sales_data.csv的文件名或内容格式例如删除文件观察系统是否会抛出错误以及错误信息是否被捕获并记录在任务输出中。这个简单的示例验证了多智能体协作框架的核心价值你只需要提出一个复合型目标系统就能自动将其拆解、分派、执行并整合最终交付一个完整的结果。7. 常见问题与排查思路在实际使用中你可能会遇到以下问题。下表列出了常见现象、原因及解决方法。问题现象可能原因排查方式解决方案启动失败提示ModuleNotFoundError依赖库未安装或虚拟环境未激活。1. 运行pip list检查openai,langchain等包是否存在。2. 确认终端提示符前有(venv)字样。1. 激活虚拟环境source venv/bin/activate。2. 安装缺失包pip install -r requirements.txt。运行时报错AuthenticationErrorOpenAI API 密钥未设置或无效。1. 检查环境变量echo $OPENAI_API_KEY。2. 在OpenAI官网检查密钥状态和余额。1. 重新设置正确的环境变量。2. 如果使用代理可能需要配置openai.api_base。智能体陷入思考循环不调用工具系统提示词Prompt设计不佳或工具描述不清。查看verboseTrue的日志观察智能体的“思考”步骤是否在重复。1. 优化系统提示词明确指示其“必须使用工具”。2. 简化工具描述使其目的更单一明确。3. 尝试使用gpt-4模型其遵循指令和推理能力更强。任务分解不合理或路由错误Orchestrator 的规划/路由提示词不够精准。打印出_plan_tasks和_route_agent中LLM的输入和原始输出。1. 在规划提示词中提供更具体的角色描述和任务示例。2. 可以尝试使用更结构化的输出格式如JSON Schema来约束LLM。代码执行工具exec安全风险高在生产环境中任意代码执行是极度危险的。审查execute_python_code函数它几乎没有任何限制。1.绝对不要在生产环境使用此示例代码。2. 使用沙箱在Docker容器内运行代码并设置资源限制CPU、内存、网络。3. 使用受限的eval或第三方安全沙箱库如restrictedpython。处理大型项目时上下文丢失working_memory是简单字典无法处理长上下文。当任务链变长时发现智能体忘记之前的步骤。1. 使用向量数据库如Chroma, Weaviate存储和检索关键信息。2. 在提示词中精炼并总结上下文而非传递全部原始信息。API调用成本过高或速度慢每个子任务和路由决策都调用LLM次数多。统计一次完整请求的LLM调用次数和总token消耗。1. 对简单、确定性的任务用规则引擎代替LLM决策。2. 使用更小、更快的模型处理简单路由。3. 实现缓存机制对相同输入缓存LLM响应。8. 最佳实践与工程建议将多智能体系统从Demo推向生产需要遵循以下最佳实践提示词工程是核心智能体的表现90%取决于提示词。务必为每个角色Orchestrator, Subagent精心设计系统提示词明确其职责、边界和输出格式。使用少样本学习Few-shot Learning在提示词中提供优秀的输入输出示例。实施严格的工具安全沙箱这是最重要的安全红线。任何执行代码、访问文件系统、调用外部API的工具都必须运行在隔离的、资源受限的环境中。考虑使用 Docker 容器或专门的 Serverless 函数来执行不可信代码。设计可观测性系统越复杂调试越困难。必须建立完善的日志系统记录每个智能体的输入、输出、工具调用和耗时。考虑使用 OpenTelemetry 等标准进行链路追踪。控制成本与延迟缓存对常见、确定的子任务结果进行缓存。模型分级用便宜、快速的小模型如 GPT-3.5-turbo处理简单任务和路由用强大、昂贵的大模型如 GPT-4处理核心的复杂规划和创作。异步执行对于可以并行执行的独立子任务使用异步机制并发执行减少总体延迟。处理“幻觉”与错误LLM会“胡言乱语”。必须对智能体的输出进行验证和过滤。例如代码执行结果需要检查语法和运行时错误数据查询结果可以抽样验证。定义清晰的停止条件防止系统陷入无限循环。可以设置最大迭代次数、最大运行时间或当连续多个任务未产生“实质性进展”时自动停止。版本化与测试将智能体的提示词、工具定义、工作流配置进行版本控制如 Git。为关键的工作流编写自动化测试确保更新不会破坏现有功能。多智能体系统代表了AI应用开发的一个前沿方向它将大语言模型从“聊天机器人”和“代码补全工具”的角色提升为了可以自主规划并执行复杂项目的“数字员工”。Oh My Subagents 这类框架的出现极大地降低了开发者构建此类系统的门槛。通过本文的实践你已经掌握了从零构建一个多智能体协作系统的核心技能定义角色、创建工具、实现协调逻辑并整合运行。虽然我们的示例是数据分析但你可以将这套模式应用到代码生成、自动化测试、智能客服、研究报告撰写等无数场景中。真正的挑战不在于搭建原型而在于如何将其工程化、产品化。这需要你在安全性、可靠性、成本和用户体验之间找到最佳平衡点。建议从一个小而具体的业务场景开始逐步迭代让智能体团队在你的项目中真正创造价值。 多智能体协作框架实战：从原理到工程实践，解决复杂任务自动化难题
你是否曾想过让一个AI助手帮你写代码、查资料、分析数据但它却常常“卡壳”要么理解不了复杂需求要么输出的结果离题万里这背后往往不是模型能力不行而是任务拆解和执行的逻辑出了问题。单个AI智能体Agent就像一个全能的实习生什么都会一点但面对一个需要多步骤、多领域协作的复杂项目时它很容易陷入混乱。今天要介绍的Oh My Subagents正是为了解决这个痛点而生。它不是一个全新的AI模型而是一个精巧的多智能体协作框架。它的核心思想是“分而治之”将一个复杂的顶层任务自动分解成一系列由不同专业子智能体Subagent执行的子任务并通过一个中央协调器Orchestrator来管理和汇总结果。简单来说它让AI从“单打独斗”变成了“团队作战”。你只需要告诉它最终目标比如“开发一个带用户登录的待办事项Web应用”它就能自动规划出“设计数据库”、“编写后端API”、“实现前端页面”、“处理用户认证”等子任务并分派给对应的子智能体去执行最后将成果整合起来。本文将带你深入探索 Oh My Subagents。我们不仅会厘清多智能体系统的核心概念更会通过一个从零开始的完整实战项目手把手教你搭建环境、编写配置、运行智能体并剖析其背后的工作流与通信机制。更重要的是我们会揭示在实际使用中可能遇到的“坑”比如任务循环、幻觉问题、成本控制并给出经过验证的最佳实践。无论你是想探索AI应用开发新范式的开发者还是正在寻找提升自动化效率工具的技术负责人这篇文章都将提供可直接落地的解决方案。1. 这篇文章真正要解决的问题在AI应用开发中我们经常面临一个困境大语言模型LLM在单轮对话或简单指令跟随上表现出色但一旦任务变得复杂、多步骤、需要长期记忆或领域专业知识时其表现就大打折扣。开发者不得不手动编写大量的提示词Prompt工程代码预先定义好所有可能的执行路径这种“硬编码”的方式不仅繁琐而且缺乏灵活性。Oh My Subagents 瞄准的正是“复杂任务自动化”这一核心难题。它试图通过一套标准化的框架将任务分解、子智能体调度、上下文传递和结果整合这些重复性工作抽象出来。开发者不再需要关注“如何让AI一步步思考”而是专注于定义“顶级目标是什么”以及“有哪些可用的专业工具子智能体”。对于读者而言阅读本文将解决以下几个具体问题认知门槛理解多智能体系统Multi-Agent System与传统单智能体或简单函数调用的本质区别。实操空白市面上概念讲解多但完整、可运行的代码示例少。本文将提供一个端到端的可复现案例。工程化盲区多智能体系统在真实项目中如何管理状态如何保证任务不陷入死循环如何控制API调用成本这些工程细节往往是失败的关键。技术选型困惑与 LangChain、AutoGPT 等项目相比Oh My Subagents 的定位和优势是什么它更适合什么场景本文的目标读者是有一定Python基础对AI应用开发感兴趣并希望将自动化能力提升到新层次的开发者。接下来的内容我们将从原理到实践彻底拆解这个框架。2. 基础概念与核心原理在深入代码之前我们必须建立清晰的概念模型。Oh My Subagents 架构中的几个核心角色理解了它们就理解了整个系统的工作流。2.1 核心角色定义角色职责类比Orchestrator (协调器)系统的“大脑”。接收用户初始请求将其分解为子任务决定执行顺序分派给合适的 Subagent并汇总最终结果。项目经理Subagent (子智能体)系统的“四肢”。专门负责某一类具体任务如写Python代码、查询数据库、分析数据。每个Subagent都有自己的系统提示词System Prompt和可用工具。专业工程师前端、后端、DBATool (工具)Subagent 可以调用的具体函数。例如“执行Python代码”、“运行SQL查询”、“调用外部API”。这扩展了AI的能力边界。工程师手中的IDE、命令行或软件Working Memory (工作记忆)一个共享的上下文存储。Orchestrator 和 Subagents 都可以从中读取信息或写入新的发现、代码、数据结果。这是智能体间通信的桥梁。团队的共享文档或协作白板Task Queue (任务队列)由Orchestrator维护的一个待处理子任务列表。系统会循环地从队列中取出任务执行直到队列为空或达到停止条件。项目看板Kanban2.2 工作流程详解整个系统的工作流程是一个典型的“规划-执行-评估”循环可以概括为以下几步任务输入用户向Orchestrator提出一个复杂请求例如“分析CSV文件sales.csv找出销量最好的产品并用图表展示”。任务规划Orchestrator 分析该请求利用其内置的LLM能力将之分解为一系列有序的子任务并放入任务队列。示例分解[“加载并查看sales.csv文件结构” “计算每个产品的总销量” “找出销量最高的产品” “生成销量排名的柱状图”]任务执行Orchestrator 从队列中取出第一个子任务根据任务类型如“数据加载”、“计算”、“可视化”选择最匹配的Subagent如“数据分析Agent”、“代码执行Agent”。工具调用被选中的Subagent 理解子任务决定需要调用哪个Tool如“pandas.read_csv”、“matplotlib.pyplot.bar”来完成工作。调用可能成功也可能失败。结果写入与循环Subagent 将执行结果成功的数据或失败的异常信息写入Working Memory。Orchestrator 检查当前子任务是否完成然后决定是继续执行队列中的下一个任务还是基于新发现的信息创建新的子任务例如发现数据有缺失新增“处理缺失值”任务。最终汇总当所有子任务完成且没有新任务产生时Orchestrator 从 Working Memory 中提取所有关键结果整合成一份完整的、面向用户的答复。这个流程的关键在于“动态规划”。与传统脚本不同Orchestrator 可以根据执行中间结果实时调整计划这使得系统能处理一些未预见的状况。2.3 与相关技术的对比为了更精准地定位 Oh My Subagents我们将其与几个常见技术进行对比技术/框架核心模式与 Oh My Subagents 的差异LangChain提供构建LLM应用所需的丰富“组件”Models, Prompts, Chains, Agents, Tools。更像一个“工具箱”。Oh My Subagents 是在 LangChain 等底层库之上封装好的一套特定范式多智能体协作的上层框架。它用 LangChain 来实现单个Agent的能力。AutoGPT一个追求完全自治的AI应用目标是给定一个目标AI能自我循环直到完成。AutoGPT 是 Oh My Subagents 理念的一个具体实现案例但通常更重、更复杂。Oh My Subagents 作为一个框架更轻量更关注提供可定制、可观察的协作机制。简单函数调用将LLM输出解析为结构化参数调用预定函数。函数调用是单次、被动的。Oh My Subagents 是主动、持续、多轮次的协作系统具备任务生成和调度能力。简单理解如果你用 LangChain 是自己在组装一台电脑那么用 Oh My Subagents 就是直接拿到了一台设计好用于“团队协作”的专用服务器。3. 环境准备与前置条件理论清晰后我们开始动手搭建。Oh My Subagents 是一个 Python 项目因此你需要一个可用的 Python 环境。3.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。本文演示基于 macOS/Linux 命令行Windows 用户建议使用 WSL2 或 Git Bash 以获得最佳体验。Python 版本Python 3.8 至 3.11。强烈推荐使用 Python 3.10 或 3.11以获得最佳的库兼容性。Python 3.12 可能因某些依赖库尚未适配而存在风险。包管理工具pip(现代 Python 已内置)。建议使用虚拟环境。API 密钥你需要一个OpenAI API 密钥或兼容 OpenAI API 的其他大模型服务如 Azure OpenAI, 国内的一些合规大模型平台的密钥。Oh My Subagents 的核心协调与执行能力依赖于大语言模型。3.2 创建虚拟环境与安装使用虚拟环境可以隔离项目依赖避免污染系统Python环境。# 1. 克隆 Oh My Subagents 仓库 (假设项目已开源在 GitHub) git clone https://github.com/your-username/oh-my-subagents.git # 请替换为实际仓库地址 cd oh-my-subagents # 2. 创建并激活虚拟环境 (以 venv 为例) python -m venv venv # 在 macOS/Linux 上激活 source venv/bin/activate # 在 Windows (CMD) 上激活 # venv\Scripts\activate # 3. 升级 pip 并安装项目依赖 pip install --upgrade pip # 安装核心依赖。如果项目有 requirements.txt则使用 # pip install -r requirements.txt # 此处我们假设需要安装以下典型依赖 pip install openai langchain langchain-openai sqlalchemy pandas matplotlib # 注实际依赖请以项目官方文档或 requirements.txt 为准3.3 配置 API 密钥安全地管理你的API密钥不要将其硬编码在代码中。推荐使用环境变量。# 在 macOS/Linux 的终端中设置环境变量 (临时仅当前会话有效) export OPENAI_API_KEYsk-your-actual-openai-api-key-here # 在 Windows (CMD) 中设置环境变量 # set OPENAI_API_KEYsk-your-actual-openai-api-key-here # 更持久的方法将上述命令添加到你的 shell 配置文件 (~/.bashrc, ~/.zshrc, 或 ~/.bash_profile) # echo export OPENAI_API_KEYsk-your-actual-key ~/.zshrc # source ~/.zshrc在你的Python代码中可以通过os.environ来读取这个密钥。import os api_key os.environ.get(OPENAI_API_KEY) if not api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量)4. 核心流程拆解构建你的第一个智能体团队现在我们开始构建一个具体的多智能体系统。我们的目标是创建一个能自动进行数据分析和可视化的团队。团队由以下角色构成Orchestrator总指挥理解用户需求并分解任务。Data Analyst Subagent数据分析专家擅长用Pandas处理数据。Code Executor Subagent代码执行专家负责安全地运行Python代码并返回结果。Visualization Subagent可视化专家擅长用Matplotlib生成图表。4.1 步骤一定义工具Tools工具是智能体能力的延伸。我们首先定义几个基础工具。# file: my_tools.py import pandas as pd import matplotlib.pyplot as plt import io import sys from contextlib import redirect_stdout, redirect_stderr from typing import Optional, Dict, Any def read_csv_file(file_path: str) - Optional[pd.DataFrame]: 读取CSV文件并返回DataFrame。 try: df pd.read_csv(file_path) return df except Exception as e: return f读取文件失败: {e} def describe_dataframe(df: pd.DataFrame) - Dict[str, Any]: 返回DataFrame的基本描述信息。 if df is None: return {error: DataFrame 为空} info { shape: df.shape, columns: df.columns.tolist(), dtypes: df.dtypes.astype(str).to_dict(), head: df.head().to_dict(orientrecords) } return info def execute_python_code(code_snippet: str) - str: 在一个受限的安全环境中执行一段Python代码。 返回标准输出和标准错误。 output_buffer io.StringIO() error_buffer io.StringIO() try: # 重定向输出和错误 with redirect_stdout(output_buffer), redirect_stderr(error_buffer): # 限制可用的全局变量增加安全性 safe_globals { __builtins__: __builtins__, pd: pd, plt: plt, df: None # 将在执行前注入 } exec(code_snippet, safe_globals) stdout output_buffer.getvalue() stderr error_buffer.getvalue() result f执行成功。\n标准输出:\n{stdout}\n标准错误:\n{stderr} if stdout or stderr else 执行成功无输出。 except Exception as e: result f执行出错: {e} return result def create_bar_chart(data_dict: dict, title: str, xlabel: str, ylabel: str) - str: 根据提供的数据字典生成一个柱状图并保存为图片。 try: names list(data_dict.keys()) values list(data_dict.values()) plt.figure(figsize(10, 6)) plt.bar(names, values) plt.title(title) plt.xlabel(xlabel) plt.ylabel(ylabel) plt.xticks(rotation45) plt.tight_layout() save_path output_chart.png plt.savefig(save_path) plt.close() return f图表已生成并保存至: {save_path} except Exception as e: return f生成图表失败: {e}关键点execute_python_code函数使用了exec这在生产环境中存在安全风险。这里仅作演示真实场景下必须使用沙箱环境如 Docker 容器或更严格的限制。工具函数应尽量保持纯净输入输出明确便于智能体理解和调用。4.2 步骤二创建子智能体Subagents每个子智能体本质上是一个具备特定系统提示词System Prompt和工具集的 LangChain Agent。# file: my_agents.py from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate import os from my_tools import read_csv_file, describe_dataframe, execute_python_code, create_bar_chart # 初始化LLM llm ChatOpenAI( modelgpt-3.5-turbo, # 或 gpt-4根据你的API权限选择 temperature0, # 降低随机性使输出更稳定 openai_api_keyos.environ.get(OPENAI_API_KEY) ) # 1. 创建数据分析子智能体 data_analysis_tools [ Tool( nameReadCSV, funcread_csv_file, description读取一个CSV文件并返回其内容。输入应为文件的路径字符串。 ), Tool( nameDescribeData, funcdescribe_dataframe, description描述一个Pandas DataFrame的基本信息包括形状、列名、数据类型和前几行数据。输入应是一个DataFrame对象。 ) ] data_analyst_prompt PromptTemplate.from_template( 你是一个专业的数据分析师。你的任务是帮助用户理解和处理数据。 你可以使用工具来读取CSV文件和查看数据集的描述信息。 当用户询问数据相关问题时请先使用工具获取数据信息再基于信息进行回答。 保持回答专业、简洁。 当前对话上下文 {context} 问题{input} 开始思考 ) data_analyst_agent create_react_agent(llmllm, toolsdata_analysis_tools, promptdata_analyst_prompt) data_analyst_executor AgentExecutor(agentdata_analyst_agent, toolsdata_analysis_tools, verboseTrue) # 2. 创建代码执行子智能体 code_execution_tools [ Tool( nameExecutePython, funcexecute_python_code, description执行一段Python代码并返回结果。输入应是一段有效的Python代码字符串。注意代码将在当前环境执行请确保安全。 ) ] code_executor_prompt PromptTemplate.from_template( 你是一个Python代码执行专家。你的唯一任务是安全地执行用户提供的Python代码片段并返回执行结果包括输出和错误。 你不应该修改代码或解释代码除非用户明确要求。 如果代码需要特定的变量如df请确保在执行前用户或上下文已提供。 当前对话上下文 {context} 代码片段{input} 开始执行 ) code_executor_agent create_react_agent(llmllm, toolscode_execution_tools, promptcode_executor_prompt) code_executor_executor AgentExecutor(agentcode_executor_agent, toolscode_execution_tools, verboseTrue) # 3. 创建可视化子智能体 viz_tools [ Tool( nameCreateBarChart, funccreate_bar_chart, description根据提供的数据字典生成柱状图。输入应为四个参数data_dict (字典), title (字符串), xlabel (字符串), ylabel (字符串)。 ) ] viz_agent_prompt PromptTemplate.from_template( 你是一个数据可视化专家。你的任务是根据用户提供的数据和需求生成合适的图表。 目前你擅长生成柱状图。请确保用户提供了生成图表所需的所有数据通常是键值对字典和标签信息。 当前对话上下文 {context} 用户需求{input} 开始工作 ) viz_agent create_react_agent(llmllm, toolsviz_tools, promptviz_agent_prompt) viz_executor AgentExecutor(agentviz_agent, toolsviz_tools, verboseTrue)关键点我们使用 LangChain 的create_react_agent来构建基于 ReAct 推理框架的智能体。AgentExecutor是实际运行智能体的对象verboseTrue会打印详细的思考过程便于调试。每个智能体都有高度定制化的系统提示词这决定了它的“专业领域”和行为模式。4.3 步骤三实现协调器Orchestrator与工作流协调器是系统的核心。它需要具备任务分解、智能体路由和上下文管理的能力。这里我们实现一个简化版本。# file: my_orchestrator.py from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from typing import List, Dict, Any import json class SimpleOrchestrator: def __init__(self, agents: Dict[str, Any], llm): 初始化协调器。 :param agents: 一个字典键为智能体角色名值为对应的AgentExecutor实例。 :param llm: 用于任务规划和路由的LLM。 self.agents agents self.llm llm self.working_memory {} # 简化的共享内存用字典实现 self.task_queue [] # 任务队列 def _plan_tasks(self, user_input: str) - List[str]: 利用LLM将用户输入分解为子任务列表。 planning_prompt f 你是一个高级任务规划师。请将用户的复杂请求分解成一系列清晰的、可顺序执行的子任务。 每个子任务应该能被以下专业角色之一处理 {, .join(self.agents.keys())} 用户请求{user_input} 请以JSON列表的形式输出子任务例如[任务1描述, 任务2描述, ...] 只输出JSON不要有其他文字。 messages [ SystemMessage(content你是一个高效的任务分解专家。), HumanMessage(contentplanning_prompt) ] response self.llm.invoke(messages) try: tasks json.loads(response.content) return tasks if isinstance(tasks, list) else [] except json.JSONDecodeError: # 如果LLM输出不是标准JSON尝试简单分割 return [task.strip() for task in response.content.split(\n) if task.strip()] def _route_agent(self, task: str) - str: 根据任务描述决定由哪个智能体处理。 routing_prompt f 请判断以下任务最适合由哪个专业智能体处理。可选的智能体有 {json.dumps(list(self.agents.keys()), indent2)} 任务描述{task} 请只返回智能体的名称不要有任何其他解释。 messages [ SystemMessage(content你是一个智能体路由专家。), HumanMessage(contentrouting_prompt) ] response self.llm.invoke(messages) agent_name response.content.strip() return agent_name if agent_name in self.agents else list(self.agents.keys())[0] # 默认返回第一个 def run(self, user_input: str) - Dict[str, Any]: 执行主工作流。 print(f[Orchestrator] 收到用户请求: {user_input}) # 1. 规划任务 self.task_queue self._plan_tasks(user_input) print(f[Orchestrator] 生成任务队列: {self.task_queue}) final_result {original_request: user_input, subtask_results: []} # 2. 循环执行任务队列 while self.task_queue: current_task self.task_queue.pop(0) print(f\n[Orchestrator] 处理任务: {current_task}) # 2.1 路由 assigned_agent_name self._route_agent(current_task) print(f[Orchestrator] 路由给智能体: {assigned_agent_name}) # 2.2 执行 # 构建上下文信息传递给智能体 context f工作记忆: {json.dumps(self.working_memory, indent2)}\n当前任务: {current_task} try: agent_executor self.agents[assigned_agent_name] # 注意这里简化了调用实际需要根据智能体类型调整输入格式 result agent_executor.invoke({input: current_task, context: context}) output result.get(output, No output) except Exception as e: output f智能体执行出错: {e} print(f[智能体 {assigned_agent_name}] 输出: {output[:200]}...) # 截断长输出 # 2.3 记录结果并更新工作记忆 task_result { task: current_task, agent: assigned_agent_name, output: output } final_result[subtask_results].append(task_result) # 简单地将最新结果存入记忆实际应更智能地提取关键信息 self.working_memory[ftask_{len(final_result[subtask_results])}] { description: current_task, result_summary: str(output)[:500] # 存储摘要 } # 2.4 可选基于结果动态添加新任务 # 这里可以加入逻辑例如如果输出包含“数据缺失”则添加“处理缺失值”任务 # if missing in output.lower(): # new_task 处理数据中的缺失值 # self.task_queue.append(new_task) # print(f[Orchestrator] 基于结果添加新任务: {new_task}) # 3. 生成最终汇总 final_result[working_memory_final] self.working_memory print(f\n[Orchestrator] 所有任务执行完毕。) return final_result关键点这个SimpleOrchestrator是一个极简实现真实框架如 Oh My Subagents的协调器会更复杂包含更优的任务分解算法、错误处理、循环检测等。_plan_tasks和_route_agent都依赖LLM这会产生额外的API调用和延迟但灵活性极高。working_memory目前是简单的字典生产系统可能需要向量数据库或更结构化的存储来管理长期、大量的上下文。5. 完整示例运行你的多智能体系统让我们将以上所有部分组合起来并模拟一个完整的用户请求。5.1 准备数据与主程序首先创建一个示例CSV文件sales_data.csv。# file: sales_data.csv Product,Month,Sales Laptop,Jan,120 Laptop,Feb,150 Laptop,Mar,180 Mouse,Jan,300 Mouse,Feb,320 Mouse,Mar,310 Keyboard,Jan,200 Keyboard,Feb,210 Keyboard,Mar,190 Monitor,Jan,80 Monitor,Feb,90 Monitor,Mar,95然后编写主程序来启动整个系统。# file: main.py import os from my_agents import data_analyst_executor, code_executor_executor, viz_executor, llm from my_orchestrator import SimpleOrchestrator def main(): # 1. 组装智能体团队 agents { DataAnalyst: data_analyst_executor, CodeExecutor: code_executor_executor, VisualizationExpert: viz_executor, } # 2. 初始化协调器 orchestrator SimpleOrchestrator(agentsagents, llmllm) # 3. 模拟用户请求 user_request 请分析当前目录下的 sales_data.csv 文件。 我需要知道 1. 数据的基本情况有哪些列多少行。 2. 每个产品Product的总销售额Sales是多少。 3. 哪个产品的总销售额最高。 4. 为每个产品的总销售额生成一个柱状图并保存为 product_sales.png。 print(*50) print(启动多智能体协作系统...) print(*50) # 4. 运行协调器 final_report orchestrator.run(user_request) # 5. 打印最终报告摘要 print(\n *50) print(任务执行最终报告) print(*50) print(f原始请求: {final_report[original_request][:100]}...) print(f\n共执行了 {len(final_report[subtask_results])} 个子任务:) for i, task_res in enumerate(final_report[subtask_results], 1): print(f {i}. [{task_res[agent]}] {task_res[task]}) print(f 输出摘要: {task_res[output][:150]}...) # 6. 检查生成的图表文件 if os.path.exists(product_sales.png): print(f\n✅ 图表文件 product_sales.png 已生成。) elif os.path.exists(output_chart.png): print(f\n 图表文件 output_chart.png 已生成默认名称。) else: print(f\n❌ 未找到生成的图表文件请检查可视化步骤的输出。) if __name__ __main__: # 确保设置了API密钥 if not os.environ.get(OPENAI_API_KEY): print(错误请设置 OPENAI_API_KEY 环境变量。) exit(1) main()5.2 运行与观察在终端中运行你的主程序# 确保在虚拟环境中且当前目录包含所有.py文件和sales_data.csv python main.py你将看到类似以下的输出具体内容因模型随机性略有不同 启动多智能体协作系统... [Orchestrator] 收到用户请求: 请分析当前目录下的 sales_data.csv 文件... [Orchestrator] 生成任务队列: [读取并查看 sales_data.csv 文件结构, 计算每个产品的总销售额, 找出总销售额最高的产品, 生成产品总销售额的柱状图] [Orchestrator] 处理任务: 读取并查看 sales_data.csv 文件结构 [Orchestrator] 路由给智能体: DataAnalyst Entering new AgentExecutor chain... 思考我需要先读取文件。我将使用ReadCSV工具。 动作ReadCSV 动作输入sales_data.csv 观察返回了一个DataFrame对象此处为简化实际为字典表示。 思考现在我需要描述这个DataFrame。使用DescribeData工具。 动作DescribeData 动作输入上一步得到的DataFrame 观察{shape: (12, 3), columns: [Product, Month, Sales], ...} 思考我有足够的信息回答。数据有12行3列列名是Product, Month, Sales。 动作最终答案 Finished chain. [智能体 DataAnalyst] 输出: 文件sales_data.csv已成功读取。数据集包含12行和3列。列名分别为Product产品、Month月份、Sales销售额。前几行数据显示了不同产品在各个月的销售情况。... ... (后续任务执行过程类似) ... [Orchestrator] 所有任务执行完毕。 任务执行最终报告 原始请求: 请分析当前目录下的 sales_data.csv 文件... 共执行了 4 个子任务: 1. [DataAnalyst] 读取并查看 sales_data.csv 文件结构 输出摘要: 文件sales_data.csv已成功读取。数据集包含12行和3列。列名分别为Product产品、Month月份、Sales销售额... 2. [CodeExecutor] 计算每个产品的总销售额 输出摘要: 执行成功。标准输出: 产品总销售额: Laptop 450, Mouse 930, Keyboard 600, Monitor 265... 3. [DataAnalyst] 找出总销售额最高的产品 输出摘要: 根据计算出的总销售额数据总销售额最高的产品是Mouse总销售额为930... 4. [VisualizationExpert] 生成产品总销售额的柱状图 输出摘要: 图表已生成并保存至: product_sales.png... ✅ 图表文件 product_sales.png 已生成。6. 运行结果与效果验证运行成功后你应该在项目目录下看到新生成的product_sales.png图表文件。用图片查看器打开应该能看到一个清晰的柱状图展示了 Laptop, Mouse, Keyboard, Monitor 四个产品的总销售额对比。如何验证系统工作正常流程完整性检查控制台输出确认4个子任务都被识别、路由并执行。结果正确性手动计算sales_data.csv中每个产品的 Sales 总和与智能体输出的结果Mouse: 930核对。打开生成的 PNG 文件确认图表标题、坐标轴标签和数据与预期一致。工作记忆在final_report变量中查看working_memory_final字段确认每个任务的结果摘要都被正确存储。错误处理你可以尝试修改sales_data.csv的文件名或内容格式例如删除文件观察系统是否会抛出错误以及错误信息是否被捕获并记录在任务输出中。这个简单的示例验证了多智能体协作框架的核心价值你只需要提出一个复合型目标系统就能自动将其拆解、分派、执行并整合最终交付一个完整的结果。7. 常见问题与排查思路在实际使用中你可能会遇到以下问题。下表列出了常见现象、原因及解决方法。问题现象可能原因排查方式解决方案启动失败提示ModuleNotFoundError依赖库未安装或虚拟环境未激活。1. 运行pip list检查openai,langchain等包是否存在。2. 确认终端提示符前有(venv)字样。1. 激活虚拟环境source venv/bin/activate。2. 安装缺失包pip install -r requirements.txt。运行时报错AuthenticationErrorOpenAI API 密钥未设置或无效。1. 检查环境变量echo $OPENAI_API_KEY。2. 在OpenAI官网检查密钥状态和余额。1. 重新设置正确的环境变量。2. 如果使用代理可能需要配置openai.api_base。智能体陷入思考循环不调用工具系统提示词Prompt设计不佳或工具描述不清。查看verboseTrue的日志观察智能体的“思考”步骤是否在重复。1. 优化系统提示词明确指示其“必须使用工具”。2. 简化工具描述使其目的更单一明确。3. 尝试使用gpt-4模型其遵循指令和推理能力更强。任务分解不合理或路由错误Orchestrator 的规划/路由提示词不够精准。打印出_plan_tasks和_route_agent中LLM的输入和原始输出。1. 在规划提示词中提供更具体的角色描述和任务示例。2. 可以尝试使用更结构化的输出格式如JSON Schema来约束LLM。代码执行工具exec安全风险高在生产环境中任意代码执行是极度危险的。审查execute_python_code函数它几乎没有任何限制。1.绝对不要在生产环境使用此示例代码。2. 使用沙箱在Docker容器内运行代码并设置资源限制CPU、内存、网络。3. 使用受限的eval或第三方安全沙箱库如restrictedpython。处理大型项目时上下文丢失working_memory是简单字典无法处理长上下文。当任务链变长时发现智能体忘记之前的步骤。1. 使用向量数据库如Chroma, Weaviate存储和检索关键信息。2. 在提示词中精炼并总结上下文而非传递全部原始信息。API调用成本过高或速度慢每个子任务和路由决策都调用LLM次数多。统计一次完整请求的LLM调用次数和总token消耗。1. 对简单、确定性的任务用规则引擎代替LLM决策。2. 使用更小、更快的模型处理简单路由。3. 实现缓存机制对相同输入缓存LLM响应。8. 最佳实践与工程建议将多智能体系统从Demo推向生产需要遵循以下最佳实践提示词工程是核心智能体的表现90%取决于提示词。务必为每个角色Orchestrator, Subagent精心设计系统提示词明确其职责、边界和输出格式。使用少样本学习Few-shot Learning在提示词中提供优秀的输入输出示例。实施严格的工具安全沙箱这是最重要的安全红线。任何执行代码、访问文件系统、调用外部API的工具都必须运行在隔离的、资源受限的环境中。考虑使用 Docker 容器或专门的 Serverless 函数来执行不可信代码。设计可观测性系统越复杂调试越困难。必须建立完善的日志系统记录每个智能体的输入、输出、工具调用和耗时。考虑使用 OpenTelemetry 等标准进行链路追踪。控制成本与延迟缓存对常见、确定的子任务结果进行缓存。模型分级用便宜、快速的小模型如 GPT-3.5-turbo处理简单任务和路由用强大、昂贵的大模型如 GPT-4处理核心的复杂规划和创作。异步执行对于可以并行执行的独立子任务使用异步机制并发执行减少总体延迟。处理“幻觉”与错误LLM会“胡言乱语”。必须对智能体的输出进行验证和过滤。例如代码执行结果需要检查语法和运行时错误数据查询结果可以抽样验证。定义清晰的停止条件防止系统陷入无限循环。可以设置最大迭代次数、最大运行时间或当连续多个任务未产生“实质性进展”时自动停止。版本化与测试将智能体的提示词、工具定义、工作流配置进行版本控制如 Git。为关键的工作流编写自动化测试确保更新不会破坏现有功能。多智能体系统代表了AI应用开发的一个前沿方向它将大语言模型从“聊天机器人”和“代码补全工具”的角色提升为了可以自主规划并执行复杂项目的“数字员工”。Oh My Subagents 这类框架的出现极大地降低了开发者构建此类系统的门槛。通过本文的实践你已经掌握了从零构建一个多智能体协作系统的核心技能定义角色、创建工具、实现协调逻辑并整合运行。虽然我们的示例是数据分析但你可以将这套模式应用到代码生成、自动化测试、智能客服、研究报告撰写等无数场景中。真正的挑战不在于搭建原型而在于如何将其工程化、产品化。这需要你在安全性、可靠性、成本和用户体验之间找到最佳平衡点。建议从一个小而具体的业务场景开始逐步迭代让智能体团队在你的项目中真正创造价值。