最近在技术圈里关于“AI Agent”智能体和“多智能体协同”的讨论热度居高不下。很多开发者都好奇当多个具备不同能力的AI智能体能够像团队一样协作时究竟能解决哪些复杂的实际问题虽然我们无法接触到企业内部正在测试的尖端平台但基于当前开源生态和主流框架完全可以在本地搭建一个简易的“多智能体协同”原型系统体验其核心工作流。本文将带你从零开始使用 Python 和 LangChain 框架构建一个模拟的“多智能体办公协同”场景。我们将创建三个具备不同角色的智能体分析师、程序员、测试员让它们围绕一个用户需求如“开发一个简单的待办事项API”进行对话与协作最终交付可运行的代码和测试报告。通过这个实战项目你将掌握智能体定义、工具调用、记忆管理以及多智能体通信的核心概念与实现。1. 背景与核心概念什么是多智能体协同在深入代码之前我们有必要厘清几个关键概念这有助于理解我们正在构建的是什么。智能体 (Agent)在AI语境下智能体通常指一个能够感知环境、进行决策并执行动作以达成目标的程序实体。一个基础的AI智能体通常由几个部分组成核心大脑 (LLM): 大型语言模型负责理解、推理和规划。工具 (Tools): 智能体可以调用的外部函数如执行代码、搜索网络、读写文件等扩展了其能力边界。记忆 (Memory): 存储智能体与用户或环境的交互历史使其具备上下文感知能力。执行器 (Executor): 协调LLM、工具和记忆决定下一步行动的逻辑。单智能体 vs. 多智能体系统单智能体系统所有任务由一个“全能”的智能体尝试解决。它可能需要调用非常多的工具处理逻辑会变得复杂且容易在需要多领域专家知识的任务上表现不佳。多智能体系统 (Multi-Agent System, MAS)由多个 specialized专业化的智能体组成。每个智能体被赋予特定的角色如“前端开发”、“数据库管理员”、“测试工程师”并配备与之相关的工具和知识。智能体之间通过某种通信机制如共享工作区、直接消息进行协作共同完成一个复杂任务。这更贴近人类团队的工作模式。“协同交付”在工作流中的体现在我们的模拟场景中“协同交付”意味着需求分发一个“主管”智能体或用户将复杂任务拆解。角色化执行不同智能体领取自己擅长的子任务。信息交换与集成智能体将各自的工作成果如分析文档、代码片段、测试用例提交到一个共享上下文中。最终组装与验证由某个智能体或最终流程将各部分成果整合形成可交付的最终产物。接下来我们将动手搭建这样一个系统的简化版本。2. 环境准备与版本说明本项目基于 Python 生态主要使用LangChain框架来简化智能体的构建过程。请确保你的开发环境满足以下要求。2.1 基础环境操作系统: Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文演示在 macOS/Linux 环境下进行Windows 用户请注意命令的细微差别如使用dir而非ls。Python 版本:Python 3.8 - 3.11。LangChain 对 3.12 的兼容性可能因依赖包而异建议使用 3.10 或 3.11 以获得最佳稳定性。包管理工具:pip(建议版本 21.0)。2.2 核心依赖库我们将创建一个新的虚拟环境并安装必要的包。首先创建项目目录并进入mkdir multi-agent-office cd multi-agent-office python -m venv venv # 激活虚拟环境 # Windows (cmd/PowerShell): venv\Scripts\activate # macOS/Linux: source venv/bin/activate激活虚拟环境后安装以下依赖。请注意openai库用于调用 GPT 模型你需要一个有效的 OpenAI API Key。如果你希望使用开源模型如通过 Ollama 部署的本地模型安装方式会有所不同本文以 OpenAI 接口为例。pip install langchain0.1.0 langchain-openai0.0.5 langchain-community0.0.10 pip install openai1.3.0 # 用于示例中运行生成的代码 pip install fastapi0.104.1 uvicorn0.24.0 pytest7.4.3版本兼容性说明LangChain 版本迭代较快以上版本组合在撰写本文时经过测试可用。如果遇到问题可以尝试固定这些版本。使用本地模型如 Ollama Llama2需要安装langchain-ollama并配置不同的 LLM 对象核心的多智能体协作逻辑是相通的。2.3 项目结构预览完成代码编写后你的项目结构将大致如下multi-agent-office/ ├── venv/ # Python 虚拟环境忽略 ├── .env # 存储环境变量如API Key ├── main.py # 主程序定义和运行多智能体 ├── shared_workspace.py # 模拟共享工作区的模块 ├── tools/ # 自定义工具目录 │ └── code_tools.py # 代码执行、文件读写等工具 └── outputs/ # 智能体协作输出的文件 ├── requirements.txt ├── spec.md ├── app.py └── test_app.py2.4 配置 API Key在项目根目录下创建.env文件用于安全地存储你的 OpenAI API Key。# .env 文件内容 OPENAI_API_KEY你的实际API密钥重要安全提示切勿将.env文件提交到 Git 等版本控制系统。确保它在.gitignore文件中。3. 核心组件拆解智能体、工具与记忆在 LangChain 中构建一个智能体本质上是组装上述几个核心组件。我们先来逐一了解它们在本项目中的实现方式。3.1 语言模型 (LLM) 初始化我们将使用 GPT-3.5-turbo 作为智能体的“大脑”它在性价比和推理能力之间取得了良好平衡。在main.py中我们首先进行初始化。# main.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 加载 .env 文件中的环境变量 load_dotenv() # 初始化LLM # temperature 控制创造性对于需要稳定输出的协作任务设置较低值 llm ChatOpenAI( modelgpt-3.5-turbo-1106, # 指定模型版本确保功能稳定 temperature0.2, api_keyos.getenv(OPENAI_API_KEY) # 从环境变量读取Key )3.2 自定义工具 (Tools) 开发工具是智能体能力的延伸。我们将创建几个简单的工具模拟办公场景中的常见操作。# tools/code_tools.py import subprocess import sys import os from typing import Type from pydantic import BaseModel, Field from langchain.tools import BaseTool class WriteFileInput(BaseModel): 写入文件的输入参数模型。 filepath: str Field(description要写入的文件路径例如outputs/app.py) content: str Field(description要写入文件的完整内容) class WriteFileTool(BaseTool): name write_file description 将给定的内容写入到指定的文件路径。如果文件已存在会被覆盖。 args_schema: Type[BaseModel] WriteFileInput def _run(self, filepath: str, content: str) - str: try: # 确保输出目录存在 os.makedirs(os.path.dirname(filepath), exist_okTrue) with open(filepath, w, encodingutf-8) as f: f.write(content) return f成功将内容写入文件{filepath} except Exception as e: return f写入文件时出错{str(e)} class RunPythonCodeInput(BaseModel): 运行Python代码的输入参数模型。 code: str Field(description需要执行的完整Python代码字符串) class RunPythonCodeTool(BaseTool): name run_python_code description 在一个安全的临时环境中执行一段Python代码并返回其输出或错误信息。适用于验证代码片段。 args_schema: Type[BaseModel] RunPythonCodeInput def _run(self, code: str) - str: try: # 使用subprocess在独立环境中运行代码避免影响主程序 result subprocess.run( [sys.executable, -c, code], capture_outputTrue, textTrue, timeout30 ) if result.returncode 0: return f代码执行成功\n{result.stdout} else: return f代码执行出错\n{result.stderr} except subprocess.TimeoutExpired: return 错误代码执行超时30秒。 except Exception as e: return f执行过程中发生未知错误{str(e)}工具设计要点输入验证使用Pydantic模型定义输入参数LLM能更好地理解如何调用工具。描述清晰description字段至关重要它帮助LLM判断在什么情况下应该调用此工具。安全性RunPythonCodeTool使用了subprocess进行沙箱化执行并设置了超时防止恶意或死循环代码影响主机。3.3 共享工作区 (Shared Workspace) - 一种记忆形式多智能体需要共享信息。我们可以用一个简单的类来模拟一个共享工作区智能体可以在这里“张贴”他们的工作成果。# shared_workspace.py class SharedWorkspace: 一个简单的共享工作区用于存储多智能体协作的中间产物。 def __init__(self): self.artifacts { requirements: None, # 存放需求分析文档 design_spec: None, # 存放设计规格说明书 source_code: None, # 存放源代码 test_cases: None, # 存放测试用例和结果 meeting_notes: [] # 存放智能体间的对话或会议纪要 } def post_artifact(self, artifact_type: str, content: str, agent_name: str): 将一个产物发布到工作区。 if artifact_type in self.artifacts: if isinstance(self.artifacts[artifact_type], list): self.artifacts[artifact_type].append(f[{agent_name}]: {content}) else: self.artifacts[artifact_type] f[{agent_name}]: {content} return f智能体 {agent_name} 成功将内容发布到 {artifact_type}。 else: # 如果类型不存在则动态创建一个列表来存储 self.artifacts[artifact_type] [f[{agent_name}]: {content}] return f创建了新板块 {artifact_type} 并发布了内容。 def get_artifact(self, artifact_type: str) - str: 从工作区获取指定类型的产物。 content self.artifacts.get(artifact_type) if content is None: return f工作区中尚未有关于 {artifact_type} 的内容。 if isinstance(content, list): return \n.join(content) else: return content def get_status(self) - str: 获取工作区当前状态概览。 status_lines [ 共享工作区状态 ] for key, value in self.artifacts.items(): if value: if isinstance(value, list): status_lines.append(f- {key}: 有 {len(value)} 条记录) else: status_lines.append(f- {key}: 已更新) else: status_lines.append(f- {key}: (空)) return \n.join(status_lines) # 创建一个全局共享工作区实例 workspace SharedWorkspace()4. 完整实战构建三智能体协作系统现在我们将把组件组装起来创建三个具有明确分工的智能体并设计它们的协作流程。4.1 定义智能体角色与系统提示词每个智能体都需要一个清晰的“人设”和职责描述这通过系统提示词 (System Prompt) 来实现。# main.py (续) from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from tools.code_tools import WriteFileTool, RunPythonCodeTool from shared_workspace import workspace # 1. 业务分析师智能体 (Business Analyst Agent) analyst_system_prompt 你是一名专业的业务分析师(BA)。你的职责是 1. 理解用户模糊或复杂的需求将其转化为清晰、可执行的技术需求规格说明。 2. 将需求拆解为具体的功能点和非功能点。 3. 将分析结果发布到共享工作区的 design_spec 板块。 4. 你可以与程序员和测试员沟通澄清需求细节。 你拥有以下工具 - write_file: 将需求文档写入文件。 - 访问共享工作区 (get_artifact, post_artifact, get_status)。 请专注于需求分析和文档撰写不要编写代码。 当前共享工作区状态 {workspace_status} # 2. 程序员智能体 (Programmer Agent) programmer_system_prompt 你是一名经验丰富的后端程序员擅长使用 Python 和 FastAPI。 你的职责是 1. 根据共享工作区中 design_spec 的需求文档编写高质量、可运行的代码。 2. 代码应包含必要的注释、错误处理和符合PEP 8规范。 3. 将编写好的源代码发布到共享工作区的 source_code 板块。 4. 可以运行 run_python_code 工具来验证代码片段的基本逻辑。 5. 可以与分析师确认需求与测试员讨论测试边界。 你拥有以下工具 - write_file: 将源代码写入文件。 - run_python_code: 运行Python代码片段进行验证。 - 访问共享工作区。 请专注于编码实现。 当前共享工作区状态 {workspace_status} # 3. 测试员智能体 (Tester Agent) tester_system_prompt 你是一名严谨的软件测试工程师。 你的职责是 1. 审查共享工作区中的 design_spec 和 source_code。 2. 编写全面的单元测试和集成测试用例确保代码覆盖主要功能点和边界情况。 3. 使用 run_python_code 工具执行测试并记录测试结果。 4. 将测试用例和测试结果发布到共享工作区的 test_cases 板块。 5. 向程序员报告发现的Bug并与分析师确认需求理解是否正确。 你拥有以下工具 - write_file: 将测试代码写入文件。 - run_python_code: 执行测试代码。 - 访问共享工作区。 请专注于测试设计和执行。 当前共享工作区状态 {workspace_status} # 访问共享工作区的工具函数我们将把它们也封装成LangChain Tool from langchain.tools import tool tool def get_workspace_status(): 获取共享工作区的当前状态概览。 return workspace.get_status() tool def post_to_workspace(artifact_type: str, content: str): 将内容发布到共享工作区的指定板块。请在发布时说明你的身份在内容中体现。 # 注意这里我们需要知道是哪个智能体在调用但LangChain工具默认不传递调用者信息。 # 一种简化方案是在提示词中要求智能体在内容开头署名。 # 更复杂的方案需要自定义Agent执行流程。这里我们做简化处理。 return workspace.post_artifact(artifact_type, content, Unknown Agent) tool def read_from_workspace(artifact_type: str): 从共享工作区读取指定板块的内容。 return workspace.get_artifact(artifact_type)4.2 创建智能体执行器为每个智能体组合其专属的工具集和提示词模板。# main.py (续) # 定义各智能体的工具集 analyst_tools [WriteFileTool(), get_workspace_status, post_to_workspace, read_from_workspace] programmer_tools [WriteFileTool(), RunPythonCodeTool(), get_workspace_status, post_to_workspace, read_from_workspace] tester_tools [WriteFileTool(), RunPythonCodeTool(), get_workspace_status, post_to_workspace, read_from_workspace] # 创建提示词模板 def create_prompt_template(system_prompt: str): 创建包含系统提示、聊天历史和用户输入的提示词模板。 return ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namechat_history), # 保留对话历史的位置 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad) # 智能体思考过程的位置 ]) # 为每个智能体创建代理Agent analyst_prompt_template create_prompt_template(analyst_system_prompt) programmer_prompt_template create_prompt_template(programmer_system_prompt) tester_prompt_template create_prompt_template(tester_system_prompt) analyst_agent create_openai_tools_agent(llm, analyst_tools, analyst_prompt_template) programmer_agent create_openai_tools_agent(llm, programmer_tools, programmer_prompt_template) tester_agent create_openai_tools_agent(llm, tester_tools, tester_prompt_template) # 创建代理执行器负责运行智能体 analyst_executor AgentExecutor(agentanalyst_agent, toolsanalyst_tools, verboseTrue, handle_parsing_errorsTrue) programmer_executor AgentExecutor(agentprogrammer_agent, toolsprogrammer_tools, verboseTrue, handle_parsing_errorsTrue) tester_executor AgentExecutor(agenttester_agent, toolstester_tools, verboseTrue, handle_parsing_errorsTrue)4.3 设计并运行协作流程我们设计一个简单的顺序工作流分析师先写需求程序员接着写代码最后测试员进行测试。在每一步智能体都会先查看工作区状态。# main.py (续) def run_collaborative_workflow(user_request: str): 运行一个简单的顺序协作工作流。 print(f\n{*60}) print(f用户需求: {user_request}) print(f{*60}\n) # 步骤 1: 业务分析师工作 print(\n 步骤1: 业务分析师开始分析需求...) analyst_response analyst_executor.invoke({ input: f用户的需求是{user_request}。请开始你的分析工作并将最终的需求规格说明书发布到共享工作区。, chat_history: [], # 初始对话历史为空 workspace_status: workspace.get_status() # 将动态状态注入提示词 }) print(f分析师输出: {analyst_response[output][:500]}...) # 截断部分输出 print(workspace.get_status()) # 步骤 2: 程序员工作 print(\n 步骤2: 程序员开始编写代码...) programmer_response programmer_executor.invoke({ input: 请根据共享工作区中的需求规格说明书 (design_spec)编写实现代码。完成后将源代码发布到工作区。, chat_history: [], workspace_status: workspace.get_status() }) print(f程序员输出: {programmer_response[output][:500]}...) print(workspace.get_status()) # 步骤 3: 测试员工作 print(\n 步骤3: 测试员开始编写并执行测试...) tester_response tester_executor.invoke({ input: 请根据共享工作区中的需求规格说明书 (design_spec) 和源代码 (source_code)编写并执行测试用例。将测试代码和结果发布到工作区。, chat_history: [], workspace_status: workspace.get_status() }) print(f测试员输出: {tester_response[output][:500]}...) print(workspace.get_status()) # 最终交付物展示 print(f\n{*60}) print(最终交付物概览) print(f{*60}) for artifact_type in [design_spec, source_code, test_cases]: content workspace.get_artifact(artifact_type) if content and content ! f工作区中尚未有关于 {artifact_type} 的内容。: print(f\n--- {artifact_type.upper()} ---\n) # 只打印前1000字符避免刷屏 print(content[:1000] (... if len(content) 1000 else )) print(f\n{*60}) print(协作流程结束。生成的文件已保存在 outputs/ 目录下。) print(f{*60}) # 主程序入口 if __name__ __main__: # 模拟一个用户需求 user_request 开发一个简单的待办事项(Todo) RESTful API需要包含创建任务、列出所有任务、根据ID获取任务、更新任务状态完成/未完成、删除任务等基本功能。使用Python和FastAPI框架数据可以暂时存储在内存中。 run_collaborative_workflow(user_request)4.4 运行与结果验证在终端中运行你的主程序python main.py你将看到类似以下的输出具体内容因模型随机性而异 用户需求: 开发一个简单的待办事项(Todo) RESTful API... 步骤1: 业务分析师开始分析需求... [分析师的一系列思考和工具调用日志...] 分析师输出: 我已将需求分析文档发布到共享工作区的 design_spec 板块... 共享工作区状态 - requirements: (空) - design_spec: 已更新 - source_code: (空) - test_cases: (空) - meeting_notes: (空) 步骤2: 程序员开始编写代码... [程序员的一系列思考和工具调用日志包括编写app.py、可能运行简单验证等...] 程序员输出: 我已根据需求编写了FastAPI应用代码并发布到 source_code... 共享工作区状态 - requirements: (空) - design_spec: 已更新 - source_code: 已更新 - test_cases: (空) - meeting_notes: (空) 步骤3: 测试员开始编写并执行测试... [测试员查看需求、代码、编写测试、运行测试、发布结果...] 测试员输出: 我已编写并执行了测试用例所有测试均已通过。结果已发布到 test_cases... 共享工作区状态 - requirements: (空) - design_spec: 已更新 - source_code: 已更新 - test_cases: 已更新 - meeting_notes: (空) 最终交付物概览 --- DESIGN_SPEC --- [分析师]: # 待办事项API需求规格说明书... 1. 功能需求... ... --- SOURCE_CODE --- [程序员]: # app.py from fastapi import FastAPI, HTTPException... app FastAPI() todos [] ... --- TEST_CASES --- [测试员]: # test_app.py import pytest from fastapi.testclient import TestClient ... 测试结果5 passed in 0.05s ... 协作流程结束。生成的文件已保存在 outputs/ 目录下。 检查outputs/目录你应该能看到生成的spec.md、app.py和test_app.py等文件。你可以手动运行python outputs/app.py启动API并在另一个终端用pytest outputs/test_app.py运行测试验证整个交付物的可运行性。5. 常见问题与排查思路在构建和运行多智能体系统时你可能会遇到以下典型问题。问题现象可能原因排查与解决思路ModuleNotFoundError: No module named langchain_openai依赖包未正确安装或版本冲突。1. 确认虚拟环境已激活。2. 运行pip list | grep langchain检查包版本。3. 严格使用pip install langchain0.1.0 langchain-openai0.0.5安装指定版本。4. 如果使用其他模型如Azure OpenAI需安装langchain-azure-openai。AuthenticationError: Incorrect API key providedOpenAI API Key 配置错误。1. 检查.env文件是否存在且OPENAI_API_KEY的值正确无误不要有多余空格。2. 在Python中print(os.getenv(OPENAI_API_KEY))确认能读取到。3. 确认API Key是否有余额或权限。智能体不调用工具一直在“空想”1. 工具描述 (description) 不清晰。2. LLM的temperature设置过高导致输出不稳定。3. 系统提示词未明确指令其使用工具。1. 检查工具的描述是否准确说明了功能和适用场景。2. 将temperature调低如0.1-0.3。3. 在系统提示词中强调“你必须使用提供的工具来完成工作”。4. 启用verboseTrue观察智能体的思考链。智能体调用工具时参数格式错误1. 工具的args_schema定义不准确。2. LLM未能正确解析用户输入为工具参数。1. 确保args_schema中的Field(description...)描述详尽。2. 使用handle_parsing_errorsTrue让执行器能处理解析错误并给LLM反馈。3. 考虑使用更强大的模型如gpt-4进行复杂任务规划。共享工作区状态未更新1.post_to_workspace工具调用失败或内容格式不对。2. 智能体没有按照指令调用该工具。1. 在工具函数内添加print语句调试是否被调用。2. 检查发布的内容字符串是否有效。3. 简化流程先让智能体用write_file工具保存到本地文件确保基础功能正常。生成的代码无法运行1. 程序员智能体生成的代码存在语法或逻辑错误。2. 依赖未声明。1. 利用RunPythonCodeTool让程序员智能体自己先验证关键代码片段。2. 在需求中明确要求生成requirements.txt。3. 这是当前AI编码的普遍局限需要人工复核和调试。6. 最佳实践与工程建议将多智能体协同从演示原型推向更稳定、可用的系统需要考虑以下工程化实践。6.1 智能体设计原则单一职责每个智能体应聚焦一个明确的领域如前端、后端、数据库设计、测试避免提示词过于复杂。能力边界清晰通过工具集严格定义智能体能做什么、不能做什么。不要给测试员智能体提供部署服务器的工具。人设与语气在系统提示词中定义智能体的“性格”如“严谨”、“有创造力”、“注重细节”这能影响其输出风格使协作更拟人化。6.2 通信与协作机制优化超越简单顺序流本文演示的是顺序工作流。更复杂的系统可以采用发布-订阅模式智能体将成果发布到特定频道关心该频道的其他智能体自动接收。管理者-工作者模式一个“主管”智能体负责任务拆解、分配和结果汇总。辩论与投票机制多个智能体对某个方案进行辩论由另一个智能体或规则进行裁决。结构化共享状态使用数据库如SQLite或向量数据库替代内存中的字典以持久化存储协作历史、中间产物和对话记录方便追溯和复盘。6.3 提示词工程动态上下文注入像本例一样将共享工作区状态、其他智能体的最新输出等作为变量注入系统提示词使智能体保持信息同步。思维链 (Chain-of-Thought) 鼓励在提示词中要求智能体“逐步思考”并利用AgentExecutor的verbose模式观察其推理过程便于调试。示例驱动 (Few-Shot)在提示词中提供一两个工具调用的正确示例能显著提高智能体使用工具的准确性。6.4 稳定性与成本控制错误处理与重试为AgentExecutor配置max_iterations和max_execution_time防止智能体陷入死循环。实现工具调用的重试逻辑。令牌 (Token) 管理多轮对话和长上下文会消耗大量Token。定期清理无关的chat_history或使用ConversationSummaryMemory等记忆组件来压缩历史。模型选型对规划、拆解等复杂任务使用能力更强的模型如GPT-4对简单的工具调用、格式转换可使用轻量级模型以平衡效果与成本。6.5 安全与伦理考量工具沙箱化任何执行代码、访问文件系统、调用外部API的工具都必须运行在严格的沙箱环境中限制其权限和资源CPU、内存、网络。输入输出审查对用户初始输入和智能体生成的最终输出进行内容安全过滤防止生成恶意代码或不当内容。人类监督在关键节点如代码合并、部署指令生成设置“人工审批”环节确保AI的决策处于人类控制之下。通过这个实战项目我们不仅复现了一个多智能体协同的雏形更深入理解了其背后的组件化思想。从定义角色、装配工具、设计工作流到优化实践每一步都是构建更复杂AI应用系统的基石。你可以在此基础上尝试引入更多智能体如运维、UI设计师设计更灵活的通信协议或集成更强大的工具如Git操作、Docker构建探索AI协同办公的更多可能性。 基于LangChain与Python构建多智能体协同办公系统实战指南
最近在技术圈里关于“AI Agent”智能体和“多智能体协同”的讨论热度居高不下。很多开发者都好奇当多个具备不同能力的AI智能体能够像团队一样协作时究竟能解决哪些复杂的实际问题虽然我们无法接触到企业内部正在测试的尖端平台但基于当前开源生态和主流框架完全可以在本地搭建一个简易的“多智能体协同”原型系统体验其核心工作流。本文将带你从零开始使用 Python 和 LangChain 框架构建一个模拟的“多智能体办公协同”场景。我们将创建三个具备不同角色的智能体分析师、程序员、测试员让它们围绕一个用户需求如“开发一个简单的待办事项API”进行对话与协作最终交付可运行的代码和测试报告。通过这个实战项目你将掌握智能体定义、工具调用、记忆管理以及多智能体通信的核心概念与实现。1. 背景与核心概念什么是多智能体协同在深入代码之前我们有必要厘清几个关键概念这有助于理解我们正在构建的是什么。智能体 (Agent)在AI语境下智能体通常指一个能够感知环境、进行决策并执行动作以达成目标的程序实体。一个基础的AI智能体通常由几个部分组成核心大脑 (LLM): 大型语言模型负责理解、推理和规划。工具 (Tools): 智能体可以调用的外部函数如执行代码、搜索网络、读写文件等扩展了其能力边界。记忆 (Memory): 存储智能体与用户或环境的交互历史使其具备上下文感知能力。执行器 (Executor): 协调LLM、工具和记忆决定下一步行动的逻辑。单智能体 vs. 多智能体系统单智能体系统所有任务由一个“全能”的智能体尝试解决。它可能需要调用非常多的工具处理逻辑会变得复杂且容易在需要多领域专家知识的任务上表现不佳。多智能体系统 (Multi-Agent System, MAS)由多个 specialized专业化的智能体组成。每个智能体被赋予特定的角色如“前端开发”、“数据库管理员”、“测试工程师”并配备与之相关的工具和知识。智能体之间通过某种通信机制如共享工作区、直接消息进行协作共同完成一个复杂任务。这更贴近人类团队的工作模式。“协同交付”在工作流中的体现在我们的模拟场景中“协同交付”意味着需求分发一个“主管”智能体或用户将复杂任务拆解。角色化执行不同智能体领取自己擅长的子任务。信息交换与集成智能体将各自的工作成果如分析文档、代码片段、测试用例提交到一个共享上下文中。最终组装与验证由某个智能体或最终流程将各部分成果整合形成可交付的最终产物。接下来我们将动手搭建这样一个系统的简化版本。2. 环境准备与版本说明本项目基于 Python 生态主要使用LangChain框架来简化智能体的构建过程。请确保你的开发环境满足以下要求。2.1 基础环境操作系统: Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文演示在 macOS/Linux 环境下进行Windows 用户请注意命令的细微差别如使用dir而非ls。Python 版本:Python 3.8 - 3.11。LangChain 对 3.12 的兼容性可能因依赖包而异建议使用 3.10 或 3.11 以获得最佳稳定性。包管理工具:pip(建议版本 21.0)。2.2 核心依赖库我们将创建一个新的虚拟环境并安装必要的包。首先创建项目目录并进入mkdir multi-agent-office cd multi-agent-office python -m venv venv # 激活虚拟环境 # Windows (cmd/PowerShell): venv\Scripts\activate # macOS/Linux: source venv/bin/activate激活虚拟环境后安装以下依赖。请注意openai库用于调用 GPT 模型你需要一个有效的 OpenAI API Key。如果你希望使用开源模型如通过 Ollama 部署的本地模型安装方式会有所不同本文以 OpenAI 接口为例。pip install langchain0.1.0 langchain-openai0.0.5 langchain-community0.0.10 pip install openai1.3.0 # 用于示例中运行生成的代码 pip install fastapi0.104.1 uvicorn0.24.0 pytest7.4.3版本兼容性说明LangChain 版本迭代较快以上版本组合在撰写本文时经过测试可用。如果遇到问题可以尝试固定这些版本。使用本地模型如 Ollama Llama2需要安装langchain-ollama并配置不同的 LLM 对象核心的多智能体协作逻辑是相通的。2.3 项目结构预览完成代码编写后你的项目结构将大致如下multi-agent-office/ ├── venv/ # Python 虚拟环境忽略 ├── .env # 存储环境变量如API Key ├── main.py # 主程序定义和运行多智能体 ├── shared_workspace.py # 模拟共享工作区的模块 ├── tools/ # 自定义工具目录 │ └── code_tools.py # 代码执行、文件读写等工具 └── outputs/ # 智能体协作输出的文件 ├── requirements.txt ├── spec.md ├── app.py └── test_app.py2.4 配置 API Key在项目根目录下创建.env文件用于安全地存储你的 OpenAI API Key。# .env 文件内容 OPENAI_API_KEY你的实际API密钥重要安全提示切勿将.env文件提交到 Git 等版本控制系统。确保它在.gitignore文件中。3. 核心组件拆解智能体、工具与记忆在 LangChain 中构建一个智能体本质上是组装上述几个核心组件。我们先来逐一了解它们在本项目中的实现方式。3.1 语言模型 (LLM) 初始化我们将使用 GPT-3.5-turbo 作为智能体的“大脑”它在性价比和推理能力之间取得了良好平衡。在main.py中我们首先进行初始化。# main.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 加载 .env 文件中的环境变量 load_dotenv() # 初始化LLM # temperature 控制创造性对于需要稳定输出的协作任务设置较低值 llm ChatOpenAI( modelgpt-3.5-turbo-1106, # 指定模型版本确保功能稳定 temperature0.2, api_keyos.getenv(OPENAI_API_KEY) # 从环境变量读取Key )3.2 自定义工具 (Tools) 开发工具是智能体能力的延伸。我们将创建几个简单的工具模拟办公场景中的常见操作。# tools/code_tools.py import subprocess import sys import os from typing import Type from pydantic import BaseModel, Field from langchain.tools import BaseTool class WriteFileInput(BaseModel): 写入文件的输入参数模型。 filepath: str Field(description要写入的文件路径例如outputs/app.py) content: str Field(description要写入文件的完整内容) class WriteFileTool(BaseTool): name write_file description 将给定的内容写入到指定的文件路径。如果文件已存在会被覆盖。 args_schema: Type[BaseModel] WriteFileInput def _run(self, filepath: str, content: str) - str: try: # 确保输出目录存在 os.makedirs(os.path.dirname(filepath), exist_okTrue) with open(filepath, w, encodingutf-8) as f: f.write(content) return f成功将内容写入文件{filepath} except Exception as e: return f写入文件时出错{str(e)} class RunPythonCodeInput(BaseModel): 运行Python代码的输入参数模型。 code: str Field(description需要执行的完整Python代码字符串) class RunPythonCodeTool(BaseTool): name run_python_code description 在一个安全的临时环境中执行一段Python代码并返回其输出或错误信息。适用于验证代码片段。 args_schema: Type[BaseModel] RunPythonCodeInput def _run(self, code: str) - str: try: # 使用subprocess在独立环境中运行代码避免影响主程序 result subprocess.run( [sys.executable, -c, code], capture_outputTrue, textTrue, timeout30 ) if result.returncode 0: return f代码执行成功\n{result.stdout} else: return f代码执行出错\n{result.stderr} except subprocess.TimeoutExpired: return 错误代码执行超时30秒。 except Exception as e: return f执行过程中发生未知错误{str(e)}工具设计要点输入验证使用Pydantic模型定义输入参数LLM能更好地理解如何调用工具。描述清晰description字段至关重要它帮助LLM判断在什么情况下应该调用此工具。安全性RunPythonCodeTool使用了subprocess进行沙箱化执行并设置了超时防止恶意或死循环代码影响主机。3.3 共享工作区 (Shared Workspace) - 一种记忆形式多智能体需要共享信息。我们可以用一个简单的类来模拟一个共享工作区智能体可以在这里“张贴”他们的工作成果。# shared_workspace.py class SharedWorkspace: 一个简单的共享工作区用于存储多智能体协作的中间产物。 def __init__(self): self.artifacts { requirements: None, # 存放需求分析文档 design_spec: None, # 存放设计规格说明书 source_code: None, # 存放源代码 test_cases: None, # 存放测试用例和结果 meeting_notes: [] # 存放智能体间的对话或会议纪要 } def post_artifact(self, artifact_type: str, content: str, agent_name: str): 将一个产物发布到工作区。 if artifact_type in self.artifacts: if isinstance(self.artifacts[artifact_type], list): self.artifacts[artifact_type].append(f[{agent_name}]: {content}) else: self.artifacts[artifact_type] f[{agent_name}]: {content} return f智能体 {agent_name} 成功将内容发布到 {artifact_type}。 else: # 如果类型不存在则动态创建一个列表来存储 self.artifacts[artifact_type] [f[{agent_name}]: {content}] return f创建了新板块 {artifact_type} 并发布了内容。 def get_artifact(self, artifact_type: str) - str: 从工作区获取指定类型的产物。 content self.artifacts.get(artifact_type) if content is None: return f工作区中尚未有关于 {artifact_type} 的内容。 if isinstance(content, list): return \n.join(content) else: return content def get_status(self) - str: 获取工作区当前状态概览。 status_lines [ 共享工作区状态 ] for key, value in self.artifacts.items(): if value: if isinstance(value, list): status_lines.append(f- {key}: 有 {len(value)} 条记录) else: status_lines.append(f- {key}: 已更新) else: status_lines.append(f- {key}: (空)) return \n.join(status_lines) # 创建一个全局共享工作区实例 workspace SharedWorkspace()4. 完整实战构建三智能体协作系统现在我们将把组件组装起来创建三个具有明确分工的智能体并设计它们的协作流程。4.1 定义智能体角色与系统提示词每个智能体都需要一个清晰的“人设”和职责描述这通过系统提示词 (System Prompt) 来实现。# main.py (续) from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from tools.code_tools import WriteFileTool, RunPythonCodeTool from shared_workspace import workspace # 1. 业务分析师智能体 (Business Analyst Agent) analyst_system_prompt 你是一名专业的业务分析师(BA)。你的职责是 1. 理解用户模糊或复杂的需求将其转化为清晰、可执行的技术需求规格说明。 2. 将需求拆解为具体的功能点和非功能点。 3. 将分析结果发布到共享工作区的 design_spec 板块。 4. 你可以与程序员和测试员沟通澄清需求细节。 你拥有以下工具 - write_file: 将需求文档写入文件。 - 访问共享工作区 (get_artifact, post_artifact, get_status)。 请专注于需求分析和文档撰写不要编写代码。 当前共享工作区状态 {workspace_status} # 2. 程序员智能体 (Programmer Agent) programmer_system_prompt 你是一名经验丰富的后端程序员擅长使用 Python 和 FastAPI。 你的职责是 1. 根据共享工作区中 design_spec 的需求文档编写高质量、可运行的代码。 2. 代码应包含必要的注释、错误处理和符合PEP 8规范。 3. 将编写好的源代码发布到共享工作区的 source_code 板块。 4. 可以运行 run_python_code 工具来验证代码片段的基本逻辑。 5. 可以与分析师确认需求与测试员讨论测试边界。 你拥有以下工具 - write_file: 将源代码写入文件。 - run_python_code: 运行Python代码片段进行验证。 - 访问共享工作区。 请专注于编码实现。 当前共享工作区状态 {workspace_status} # 3. 测试员智能体 (Tester Agent) tester_system_prompt 你是一名严谨的软件测试工程师。 你的职责是 1. 审查共享工作区中的 design_spec 和 source_code。 2. 编写全面的单元测试和集成测试用例确保代码覆盖主要功能点和边界情况。 3. 使用 run_python_code 工具执行测试并记录测试结果。 4. 将测试用例和测试结果发布到共享工作区的 test_cases 板块。 5. 向程序员报告发现的Bug并与分析师确认需求理解是否正确。 你拥有以下工具 - write_file: 将测试代码写入文件。 - run_python_code: 执行测试代码。 - 访问共享工作区。 请专注于测试设计和执行。 当前共享工作区状态 {workspace_status} # 访问共享工作区的工具函数我们将把它们也封装成LangChain Tool from langchain.tools import tool tool def get_workspace_status(): 获取共享工作区的当前状态概览。 return workspace.get_status() tool def post_to_workspace(artifact_type: str, content: str): 将内容发布到共享工作区的指定板块。请在发布时说明你的身份在内容中体现。 # 注意这里我们需要知道是哪个智能体在调用但LangChain工具默认不传递调用者信息。 # 一种简化方案是在提示词中要求智能体在内容开头署名。 # 更复杂的方案需要自定义Agent执行流程。这里我们做简化处理。 return workspace.post_artifact(artifact_type, content, Unknown Agent) tool def read_from_workspace(artifact_type: str): 从共享工作区读取指定板块的内容。 return workspace.get_artifact(artifact_type)4.2 创建智能体执行器为每个智能体组合其专属的工具集和提示词模板。# main.py (续) # 定义各智能体的工具集 analyst_tools [WriteFileTool(), get_workspace_status, post_to_workspace, read_from_workspace] programmer_tools [WriteFileTool(), RunPythonCodeTool(), get_workspace_status, post_to_workspace, read_from_workspace] tester_tools [WriteFileTool(), RunPythonCodeTool(), get_workspace_status, post_to_workspace, read_from_workspace] # 创建提示词模板 def create_prompt_template(system_prompt: str): 创建包含系统提示、聊天历史和用户输入的提示词模板。 return ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namechat_history), # 保留对话历史的位置 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad) # 智能体思考过程的位置 ]) # 为每个智能体创建代理Agent analyst_prompt_template create_prompt_template(analyst_system_prompt) programmer_prompt_template create_prompt_template(programmer_system_prompt) tester_prompt_template create_prompt_template(tester_system_prompt) analyst_agent create_openai_tools_agent(llm, analyst_tools, analyst_prompt_template) programmer_agent create_openai_tools_agent(llm, programmer_tools, programmer_prompt_template) tester_agent create_openai_tools_agent(llm, tester_tools, tester_prompt_template) # 创建代理执行器负责运行智能体 analyst_executor AgentExecutor(agentanalyst_agent, toolsanalyst_tools, verboseTrue, handle_parsing_errorsTrue) programmer_executor AgentExecutor(agentprogrammer_agent, toolsprogrammer_tools, verboseTrue, handle_parsing_errorsTrue) tester_executor AgentExecutor(agenttester_agent, toolstester_tools, verboseTrue, handle_parsing_errorsTrue)4.3 设计并运行协作流程我们设计一个简单的顺序工作流分析师先写需求程序员接着写代码最后测试员进行测试。在每一步智能体都会先查看工作区状态。# main.py (续) def run_collaborative_workflow(user_request: str): 运行一个简单的顺序协作工作流。 print(f\n{*60}) print(f用户需求: {user_request}) print(f{*60}\n) # 步骤 1: 业务分析师工作 print(\n 步骤1: 业务分析师开始分析需求...) analyst_response analyst_executor.invoke({ input: f用户的需求是{user_request}。请开始你的分析工作并将最终的需求规格说明书发布到共享工作区。, chat_history: [], # 初始对话历史为空 workspace_status: workspace.get_status() # 将动态状态注入提示词 }) print(f分析师输出: {analyst_response[output][:500]}...) # 截断部分输出 print(workspace.get_status()) # 步骤 2: 程序员工作 print(\n 步骤2: 程序员开始编写代码...) programmer_response programmer_executor.invoke({ input: 请根据共享工作区中的需求规格说明书 (design_spec)编写实现代码。完成后将源代码发布到工作区。, chat_history: [], workspace_status: workspace.get_status() }) print(f程序员输出: {programmer_response[output][:500]}...) print(workspace.get_status()) # 步骤 3: 测试员工作 print(\n 步骤3: 测试员开始编写并执行测试...) tester_response tester_executor.invoke({ input: 请根据共享工作区中的需求规格说明书 (design_spec) 和源代码 (source_code)编写并执行测试用例。将测试代码和结果发布到工作区。, chat_history: [], workspace_status: workspace.get_status() }) print(f测试员输出: {tester_response[output][:500]}...) print(workspace.get_status()) # 最终交付物展示 print(f\n{*60}) print(最终交付物概览) print(f{*60}) for artifact_type in [design_spec, source_code, test_cases]: content workspace.get_artifact(artifact_type) if content and content ! f工作区中尚未有关于 {artifact_type} 的内容。: print(f\n--- {artifact_type.upper()} ---\n) # 只打印前1000字符避免刷屏 print(content[:1000] (... if len(content) 1000 else )) print(f\n{*60}) print(协作流程结束。生成的文件已保存在 outputs/ 目录下。) print(f{*60}) # 主程序入口 if __name__ __main__: # 模拟一个用户需求 user_request 开发一个简单的待办事项(Todo) RESTful API需要包含创建任务、列出所有任务、根据ID获取任务、更新任务状态完成/未完成、删除任务等基本功能。使用Python和FastAPI框架数据可以暂时存储在内存中。 run_collaborative_workflow(user_request)4.4 运行与结果验证在终端中运行你的主程序python main.py你将看到类似以下的输出具体内容因模型随机性而异 用户需求: 开发一个简单的待办事项(Todo) RESTful API... 步骤1: 业务分析师开始分析需求... [分析师的一系列思考和工具调用日志...] 分析师输出: 我已将需求分析文档发布到共享工作区的 design_spec 板块... 共享工作区状态 - requirements: (空) - design_spec: 已更新 - source_code: (空) - test_cases: (空) - meeting_notes: (空) 步骤2: 程序员开始编写代码... [程序员的一系列思考和工具调用日志包括编写app.py、可能运行简单验证等...] 程序员输出: 我已根据需求编写了FastAPI应用代码并发布到 source_code... 共享工作区状态 - requirements: (空) - design_spec: 已更新 - source_code: 已更新 - test_cases: (空) - meeting_notes: (空) 步骤3: 测试员开始编写并执行测试... [测试员查看需求、代码、编写测试、运行测试、发布结果...] 测试员输出: 我已编写并执行了测试用例所有测试均已通过。结果已发布到 test_cases... 共享工作区状态 - requirements: (空) - design_spec: 已更新 - source_code: 已更新 - test_cases: 已更新 - meeting_notes: (空) 最终交付物概览 --- DESIGN_SPEC --- [分析师]: # 待办事项API需求规格说明书... 1. 功能需求... ... --- SOURCE_CODE --- [程序员]: # app.py from fastapi import FastAPI, HTTPException... app FastAPI() todos [] ... --- TEST_CASES --- [测试员]: # test_app.py import pytest from fastapi.testclient import TestClient ... 测试结果5 passed in 0.05s ... 协作流程结束。生成的文件已保存在 outputs/ 目录下。 检查outputs/目录你应该能看到生成的spec.md、app.py和test_app.py等文件。你可以手动运行python outputs/app.py启动API并在另一个终端用pytest outputs/test_app.py运行测试验证整个交付物的可运行性。5. 常见问题与排查思路在构建和运行多智能体系统时你可能会遇到以下典型问题。问题现象可能原因排查与解决思路ModuleNotFoundError: No module named langchain_openai依赖包未正确安装或版本冲突。1. 确认虚拟环境已激活。2. 运行pip list | grep langchain检查包版本。3. 严格使用pip install langchain0.1.0 langchain-openai0.0.5安装指定版本。4. 如果使用其他模型如Azure OpenAI需安装langchain-azure-openai。AuthenticationError: Incorrect API key providedOpenAI API Key 配置错误。1. 检查.env文件是否存在且OPENAI_API_KEY的值正确无误不要有多余空格。2. 在Python中print(os.getenv(OPENAI_API_KEY))确认能读取到。3. 确认API Key是否有余额或权限。智能体不调用工具一直在“空想”1. 工具描述 (description) 不清晰。2. LLM的temperature设置过高导致输出不稳定。3. 系统提示词未明确指令其使用工具。1. 检查工具的描述是否准确说明了功能和适用场景。2. 将temperature调低如0.1-0.3。3. 在系统提示词中强调“你必须使用提供的工具来完成工作”。4. 启用verboseTrue观察智能体的思考链。智能体调用工具时参数格式错误1. 工具的args_schema定义不准确。2. LLM未能正确解析用户输入为工具参数。1. 确保args_schema中的Field(description...)描述详尽。2. 使用handle_parsing_errorsTrue让执行器能处理解析错误并给LLM反馈。3. 考虑使用更强大的模型如gpt-4进行复杂任务规划。共享工作区状态未更新1.post_to_workspace工具调用失败或内容格式不对。2. 智能体没有按照指令调用该工具。1. 在工具函数内添加print语句调试是否被调用。2. 检查发布的内容字符串是否有效。3. 简化流程先让智能体用write_file工具保存到本地文件确保基础功能正常。生成的代码无法运行1. 程序员智能体生成的代码存在语法或逻辑错误。2. 依赖未声明。1. 利用RunPythonCodeTool让程序员智能体自己先验证关键代码片段。2. 在需求中明确要求生成requirements.txt。3. 这是当前AI编码的普遍局限需要人工复核和调试。6. 最佳实践与工程建议将多智能体协同从演示原型推向更稳定、可用的系统需要考虑以下工程化实践。6.1 智能体设计原则单一职责每个智能体应聚焦一个明确的领域如前端、后端、数据库设计、测试避免提示词过于复杂。能力边界清晰通过工具集严格定义智能体能做什么、不能做什么。不要给测试员智能体提供部署服务器的工具。人设与语气在系统提示词中定义智能体的“性格”如“严谨”、“有创造力”、“注重细节”这能影响其输出风格使协作更拟人化。6.2 通信与协作机制优化超越简单顺序流本文演示的是顺序工作流。更复杂的系统可以采用发布-订阅模式智能体将成果发布到特定频道关心该频道的其他智能体自动接收。管理者-工作者模式一个“主管”智能体负责任务拆解、分配和结果汇总。辩论与投票机制多个智能体对某个方案进行辩论由另一个智能体或规则进行裁决。结构化共享状态使用数据库如SQLite或向量数据库替代内存中的字典以持久化存储协作历史、中间产物和对话记录方便追溯和复盘。6.3 提示词工程动态上下文注入像本例一样将共享工作区状态、其他智能体的最新输出等作为变量注入系统提示词使智能体保持信息同步。思维链 (Chain-of-Thought) 鼓励在提示词中要求智能体“逐步思考”并利用AgentExecutor的verbose模式观察其推理过程便于调试。示例驱动 (Few-Shot)在提示词中提供一两个工具调用的正确示例能显著提高智能体使用工具的准确性。6.4 稳定性与成本控制错误处理与重试为AgentExecutor配置max_iterations和max_execution_time防止智能体陷入死循环。实现工具调用的重试逻辑。令牌 (Token) 管理多轮对话和长上下文会消耗大量Token。定期清理无关的chat_history或使用ConversationSummaryMemory等记忆组件来压缩历史。模型选型对规划、拆解等复杂任务使用能力更强的模型如GPT-4对简单的工具调用、格式转换可使用轻量级模型以平衡效果与成本。6.5 安全与伦理考量工具沙箱化任何执行代码、访问文件系统、调用外部API的工具都必须运行在严格的沙箱环境中限制其权限和资源CPU、内存、网络。输入输出审查对用户初始输入和智能体生成的最终输出进行内容安全过滤防止生成恶意代码或不当内容。人类监督在关键节点如代码合并、部署指令生成设置“人工审批”环节确保AI的决策处于人类控制之下。通过这个实战项目我们不仅复现了一个多智能体协同的雏形更深入理解了其背后的组件化思想。从定义角色、装配工具、设计工作流到优化实践每一步都是构建更复杂AI应用系统的基石。你可以在此基础上尝试引入更多智能体如运维、UI设计师设计更灵活的通信协议或集成更强大的工具如Git操作、Docker构建探索AI协同办公的更多可能性。