基于LangChain与Python构建多智能体协同办公系统实战指南

最近在技术圈里，关于“AI Agent”（智能体）和“多智能体协同”的讨论热度居高不下。很多开发者都好奇，当多个具备不同能力的AI智能体能够像团队一样协作时，究竟能解决哪些复杂的实际问题？虽然我们无法接触到企业内部正在测试的尖端平台，但基于当前开源生态和主流框架，完全可以在本地搭建一个简易的“多智能体协同”原型系统，体验其核心工作流。

本文将带你从零开始，使用 Python 和 LangChain 框架，构建一个模拟的“多智能体办公协同”场景。我们将创建三个具备不同角色的智能体（分析师、程序员、测试员），让它们围绕一个用户需求（如“开发一个简单的待办事项API”）进行对话与协作，最终交付可运行的代码和测试报告。通过这个实战项目，你将掌握智能体定义、工具调用、记忆管理以及多智能体通信的核心概念与实现。

1. 背景与核心概念：什么是多智能体协同？

在深入代码之前，我们有必要厘清几个关键概念，这有助于理解我们正在构建的是什么。

智能体 (Agent)在AI语境下，智能体通常指一个能够感知环境、进行决策并执行动作以达成目标的程序实体。一个基础的AI智能体通常由几个部分组成：

1. 核心大脑 (LLM): 大型语言模型，负责理解、推理和规划。
2. 工具 (Tools): 智能体可以调用的外部函数，如执行代码、搜索网络、读写文件等，扩展了其能力边界。
3. 记忆 (Memory): 存储智能体与用户或环境的交互历史，使其具备上下文感知能力。
4. 执行器 (Executor): 协调LLM、工具和记忆，决定下一步行动的逻辑。

单智能体 vs. 多智能体系统

• 单智能体系统：所有任务由一个“全能”的智能体尝试解决。它可能需要调用非常多的工具，处理逻辑会变得复杂，且容易在需要多领域专家知识的任务上表现不佳。
• 多智能体系统 (Multi-Agent System, MAS)：由多个 specialized（专业化）的智能体组成。每个智能体被赋予特定的角色（如“前端开发”、“数据库管理员”、“测试工程师”），并配备与之相关的工具和知识。智能体之间通过某种通信机制（如共享工作区、直接消息）进行协作，共同完成一个复杂任务。这更贴近人类团队的工作模式。

“协同交付”在工作流中的体现在我们的模拟场景中，“协同交付”意味着：

1. 需求分发：一个“主管”智能体或用户将复杂任务拆解。
2. 角色化执行：不同智能体领取自己擅长的子任务。
3. 信息交换与集成：智能体将各自的工作成果（如分析文档、代码片段、测试用例）提交到一个共享上下文中。
4. 最终组装与验证：由某个智能体或最终流程，将各部分成果整合，形成可交付的最终产物。

接下来，我们将动手搭建这样一个系统的简化版本。

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 langchain==0.1.0 langchain-openai==0.0.5 langchain-community==0.0.10 pip install openai==1.3.0 # 用于示例中运行生成的代码 pip install fastapi==0.104.1 uvicorn==0.24.0 pytest==7.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.py

2.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( model="gpt-3.5-turbo-1106", # 指定模型版本，确保功能稳定 temperature=0.2, api_key=os.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_ok=True) with open(filepath, 'w', encoding='utf-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_output=True, text=True, timeout=30 ) 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)}"

工具设计要点：

1. 输入验证：使用Pydantic模型定义输入参数，LLM能更好地理解如何调用工具。
2. 描述清晰：description字段至关重要，它帮助LLM判断在什么情况下应该调用此工具。
3. 安全性：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_name="chat_history"), # 保留对话历史的位置 ("human", "{input}"), MessagesPlaceholder(variable_name="agent_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(agent=analyst_agent, tools=analyst_tools, verbose=True, handle_parsing_errors=True) programmer_executor = AgentExecutor(agent=programmer_agent, tools=programmer_tools, verbose=True, handle_parsing_errors=True) tester_executor = AgentExecutor(agent=tester_agent, tools=tester_tools, verbose=True, handle_parsing_errors=True)

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. 常见问题与排查思路

在构建和运行多智能体系统时，你可能会遇到以下典型问题。

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协同办公的更多可能性。