最近在尝试将AI Agent技术应用到实际开发流程中发现单Agent虽然能处理特定任务但在面对复杂、多步骤的软件开发场景时往往力不从心。比如从需求分析到代码生成、再到测试和部署单一模型很难兼顾所有环节的准确性和上下文一致性。这时一个能够协调多个专业Agent、实现任务自动流转的“多Agent自动化开发系统”就成了破局的关键。本文将基于“Loop Engineering”这一工程化思想手把手带你搭建一个实用的多Agent协作系统涵盖从核心概念、环境搭建、代码实战到生产级优化的完整闭环。无论你是想探索AI辅助开发的工程师还是希望提升团队研发效能的技术负责人都能从中获得可直接复用的方案。1. 背景与核心概念为什么需要多Agent开发系统在深入代码之前我们有必要厘清几个核心概念理解多Agent系统要解决的根本问题。1.1 什么是AI AgentAI Agent智能体可以理解为一个能够感知环境、进行决策并执行动作以达成目标的智能程序。在软件开发语境下一个Agent通常由一个大语言模型LLM驱动配备特定的工具Tools和记忆Memory专注于完成某一类任务例如代码生成Agent根据自然语言描述或函数签名生成代码片段。代码审查Agent检查代码风格、潜在Bug和安全漏洞。测试生成Agent根据代码逻辑自动生成单元测试用例。文档编写Agent为函数或模块生成API文档。单个Agent的能力是垂直且有限的。1.2 单Agent的局限与多Agent系统的优势当你让一个“代码生成Agent”去完成“开发一个用户登录API”的任务时它可能生成功能代码但通常不会同时考虑为这段代码编写对应的单元测试。检查生成的代码是否符合项目的安全规范如密码加密。更新相关的API接口文档。将代码集成到现有项目结构中。如果要求一个Agent“全能”其Prompt会变得极其复杂效果也难以保证。多Agent系统的核心思想是“专业的人做专业的事”。通过设计一个协调者Orchestrator或工作流Workflow将一个大任务分解并路由给最专业的Agent去执行最后汇总结果。这更贴近真实的软件工程团队协作模式。1.3 Loop Engineering从实验到工程化“Loop Engineering”强调的是将AI能力特别是多Agent协作以可预测、可管理、可复现的方式集成到现有开发流程中。它不是一个具体的框架而是一种方法论关注确定性给定相同的输入系统应产生稳定、可预期的输出。可观测性能够追踪每个Agent的决策过程、工具调用和中间状态。容错与回滚当某个Agent执行失败时系统应有备选方案或回滚机制。人机协同在关键节点如代码合并、生产部署引入人工审核。我们即将构建的系统正是Loop Engineering思想的一次实践。2. 环境准备与版本说明我们将使用Python作为主要开发语言并借助LangChain这一流行的LLM应用框架来构建Agent。选择它是因为其丰富的Agent、Tool和Chain抽象能极大简化多Agent系统的搭建。2.1 基础环境操作系统macOS / Linux (推荐) 或 Windows (WSL2)。本文命令以Linux/macOS为例。Python版本 3.9。建议使用3.10或3.11以获得最佳兼容性。包管理工具pip或poetry。本文使用pip。代码编辑器VS Code, PyCharm 等均可。2.2 核心依赖库我们将创建一个新的项目目录并安装依赖。首先确保你的环境已安装pip。# 创建项目目录 mkdir multi-agent-dev-system cd multi-agent-dev-system # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langchain langchain-openai langchain-community # 安装用于处理代码的库 pip install astor black # black用于代码格式化astor用于代码生成 # 安装用于工作流管理的库示例使用LangGraph它是LangChain的一部分用于构建有状态的多Actor应用 pip install langgraph # 安装用于项目管理的库 pip install python-dotenv # 管理环境变量2.3 LLM API 密钥配置本示例使用OpenAI的GPT模型作为Agent的“大脑”。你需要准备一个OpenAI API Key。访问 OpenAI平台 创建API Key。在项目根目录创建.env文件并填入你的密钥# .env 文件 OPENAI_API_KEYsk-your-actual-api-key-here重要请勿将.env文件提交到版本控制系统如Git。确保它在.gitignore中。2.4 项目结构预览在开始编码前我们先规划一下项目结构这有助于理解后续的代码组织。multi-agent-dev-system/ ├── .env # 环境变量配置文件 ├── .gitignore # Git忽略文件 ├── requirements.txt # 项目依赖清单可由 pip freeze requirements.txt 生成 ├── main.py # 系统主入口协调工作流 ├── agents/ # 各个Agent的实现 │ ├── __init__.py │ ├── planner.py # 任务规划Agent │ ├── coder.py # 代码生成Agent │ ├── reviewer.py # 代码审查Agent │ └── tester.py # 测试生成Agent ├── tools/ # Agent可用的工具集 │ ├── __init__.py │ ├── code_tools.py # 代码相关工具如读写文件、执行测试 │ └── system_tools.py # 系统相关工具如执行Shell命令 ├── workflows/ # 工作流定义 │ ├── __init__.py │ └── dev_workflow.py # 核心开发工作流 └── utils/ # 工具函数 ├── __init__.py └── state.py # 定义工作流状态3. 核心组件拆解Agent、Tool与State多Agent系统的基石是三个概念具有特定能力的Agent、Agent可以调用的Tool以及在不同Agent间传递信息的State。3.1 定义Agent在LangChain中一个Agent通常由一个LLM、一系列Tools和一个特定的Prompt模板构成。我们将创建几个基础Agent。首先创建agents/__init__.py并设置一个通用的Agent创建函数。# agents/__init__.py import os from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate from dotenv import load_dotenv load_dotenv() # 加载.env文件中的环境变量 def create_base_llm(temperature0.1, model_namegpt-4o-mini): 创建基础的LLM实例。 # 确保API_KEY已设置 if not os.getenv(OPENAI_API_KEY): raise ValueError(请在.env文件中设置OPENAI_API_KEY环境变量。) return ChatOpenAI( modelmodel_name, temperaturetemperature, # 温度越低输出越确定 api_keyos.getenv(OPENAI_API_KEY) ) def create_agent(llm, tools, system_prompt): 创建一个基于ReAct范式的Agent。 # ReAct提示词模板告诉Agent如何思考Reason和行动Act prompt_template PromptTemplate.from_template( f{system_prompt} 你有权限使用以下工具 {{tools}} 使用以下格式 任务你必须完成的任务 思考你需要思考如何完成任务 行动要使用的工具名称必须是[{, .join([tool.name for tool in tools])}]中的一个 行动输入工具的输入 观察工具执行的结果 ... (这个思考/行动/观察循环可以重复多次) 思考我现在知道最终答案了 最终答案对原始任务输入的回答 开始 任务{{input}} 思考{{agent_scratchpad}} ) agent create_react_agent(llmllm, toolstools, promptprompt_template) return AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue)3.2 实现工具ToolsTools是Agent与外界交互的“手”。我们实现几个开发相关的工具。# tools/code_tools.py import ast import subprocess import sys from typing import Type from langchain.tools import BaseTool, tool from pydantic import BaseModel, Field class FileReadInput(BaseModel): 读取文件的工具输入模型。 file_path: str Field(description要读取的文件的路径) class FileWriteInput(BaseModel): 写入文件的工具输入模型。 file_path: str Field(description要写入的文件的路径) content: str Field(description要写入文件的内容) class RunTestInput(BaseModel): 运行测试的工具输入模型。 test_command: str Field(description运行测试的命令例如 pytest tests/test_math.py) tool(args_schemaFileReadInput) def read_file(file_path: str) - str: 读取指定文件的内容。 try: with open(file_path, r, encodingutf-8) as f: return f.read() except FileNotFoundError: return f错误文件 {file_path} 未找到。 except Exception as e: return f读取文件时出错{str(e)} tool(args_schemaFileWriteInput) def write_file(file_path: str, content: str) - str: 将内容写入指定文件。如果文件存在则覆盖。 try: # 确保目录存在 os.makedirs(os.path.dirname(file_path), exist_okTrue) with open(file_path, w, encodingutf-8) as f: f.write(content) return f成功将内容写入文件{file_path} except Exception as e: return f写入文件时出错{str(e)} tool(args_schemaRunTestInput) def run_pytest(test_command: str) - str: 运行pytest测试命令并返回结果。 try: # 注意在生产环境中应对命令进行严格校验以防止注入攻击 result subprocess.run( test_command.split(), capture_outputTrue, textTrue, cwdos.getcwd() # 在当前项目目录下运行 ) output fSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\n返回码{result.returncode} return output except Exception as e: return f执行测试命令时出错{str(e)} # 一个更复杂的工具解析Python代码并提取函数信息 tool def analyze_python_code(code: str) - str: 分析一段Python代码返回其中的函数和类定义列表。 try: tree ast.parse(code) definitions [] for node in ast.walk(tree): if isinstance(node, ast.FunctionDef): definitions.append(f函数: {node.name} (行号: {node.lineno})) elif isinstance(node, ast.ClassDef): definitions.append(f类: {node.name} (行号: {node.lineno})) if definitions: return \n.join(definitions) else: return 未找到函数或类定义。 except SyntaxError as e: return f代码语法错误{e}3.3 定义工作流状态State在多Agent工作流中需要有一个共享的“状态”对象来在不同节点间传递信息。我们使用Pydantic模型来定义它。# utils/state.py from typing import TypedDict, List, Optional, Annotated import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): 多Agent工作流的共享状态定义。 # 原始的用户需求 requirements: str # 任务规划Agent输出的步骤列表 plan: List[str] # 当前正在执行的步骤索引 current_step: int # 代码生成Agent生成的代码 generated_code: Optional[str] # 代码文件路径 code_file_path: Optional[str] # 代码审查Agent输出的审查意见 review_feedback: Optional[str] # 测试生成Agent生成的测试代码 test_code: Optional[str] # 测试文件路径 test_file_path: Optional[str] # 测试执行结果 test_result: Optional[str] # 最终的系统输出或总结 final_output: Optional[str] # 用于存储对话历史LangGraph内置 messages: Annotated[list, add_messages]4. 完整实战构建多Agent开发工作流现在我们将使用LangGraph来编排一个完整的开发工作流。这个工作流模拟一个简化但完整的开发过程规划 - 编码 - 审查 - 测试。4.1 创建各个职能Agent首先在agents目录下创建具体的Agent。# agents/planner.py from langchain_core.prompts import ChatPromptTemplate from ..utils.state import AgentState from ..agents import create_base_llm def create_planner_agent(): 创建任务规划Agent。它不调用工具只进行思考分解。 llm create_base_llm(temperature0.1) system_prompt 你是一个资深的软件开发项目经理。你的任务是将一个模糊的软件开发需求分解成具体、可执行、顺序正确的开发步骤。 步骤应该清晰无歧义例如 1. 分析需求确定需要创建的模块和函数。 2. 编写核心业务逻辑函数X。 3. 编写相关的工具函数Y。 4. 为函数X和Y编写单元测试。 5. 运行测试并确保通过。 请直接输出步骤列表不要有多余的解释。 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, 需求{requirements}) ]) # 这是一个简单的LLMChain不是Tool-calling Agent return prompt | llm def planning_node(state: AgentState): 工作流节点执行任务规划。 print(f\n 阶段1任务规划 ) print(f原始需求{state[requirements]}) planner create_planner_agent() plan_result planner.invoke({requirements: state[requirements]}) # 假设LLM返回的是用换行符分隔的步骤字符串 plan_steps [step.strip() for step in plan_result.content.split(\n) if step.strip()] # 更新状态 state[plan] plan_steps state[current_step] 0 print(f生成计划{plan_steps}) return state# agents/coder.py from langchain_core.prompts import ChatPromptTemplate from ..utils.state import AgentState from ..agents import create_base_llm, create_agent from ..tools.code_tools import write_file, read_file, analyze_python_code def create_coder_agent(): 创建代码生成Agent。 llm create_base_llm(temperature0.2) # 编码可以稍有一点创造性 tools [write_file, read_file, analyze_python_code] system_prompt 你是一个经验丰富的Python软件工程师。根据给定的任务步骤和上下文编写高质量、可维护的Python代码。 你可以使用以下工具 - write_file: 将生成的代码写入指定文件。 - read_file: 读取现有文件以了解项目结构或上下文。 - analyze_python_code: 分析代码结构。 请确保代码符合PEP 8规范包含适当的注释和异常处理。 return create_agent(llm, tools, system_prompt) def coding_node(state: AgentState): 工作流节点执行代码生成。 print(f\n 阶段2代码生成 ) if not state[plan]: state[final_output] 错误未找到开发计划。 return state current_step state[plan][state[current_step]] print(f执行步骤{current_step}) coder create_coder_agent() # 构建给Agent的输入包含计划、当前步骤和已有代码如果有 input_for_coder f 开发计划{state[plan]} 当前需要完成的步骤{current_step} 请生成或修改代码来完成此步骤。如果需要可以读取项目中的其他文件来了解上下文。 result coder.invoke({input: input_for_coder}) state[generated_code] result.get(output, 代码生成Agent未返回结果。) # 这里简化处理假设Agent会将代码写入文件并更新文件路径到状态。 # 在实际中你需要从Agent的执行结果中解析出文件路径。 state[code_file_path] generated_code.py # 示例路径 state[current_step] 1 # 移动到下一个步骤 return state类似地创建agents/reviewer.py和agents/tester.py。由于篇幅这里给出reviewer的示例。# agents/reviewer.py from ..utils.state import AgentState from ..agents import create_base_llm, create_agent from ..tools.code_tools import read_file, analyze_python_code def create_reviewer_agent(): 创建代码审查Agent。 llm create_base_llm(temperature0) tools [read_file, analyze_python_code] system_prompt 你是一个严格的代码审查员。你的任务是检查给定代码的质量、安全性、性能和可维护性。 重点关注 1. 语法错误和潜在的运行时错误。 2. 安全漏洞如SQL注入、命令注入。 3. 代码风格是否符合PEP 8。 4. 函数和变量命名是否清晰。 5. 是否有重复代码。 6. 异常处理是否完备。 请提供具体的、可操作的修改建议。 return create_agent(llm, tools, system_prompt) def review_node(state: AgentState): 工作流节点执行代码审查。 print(f\n 阶段3代码审查 ) if not state[code_file_path]: state[review_feedback] 警告未找到生成的代码文件跳过审查。 return state reviewer create_reviewer_agent() input_for_reviewer f 请审查以下文件中的代码{state[code_file_path]} 请提供详细的审查意见。 result reviewer.invoke({input: input_for_reviewer}) state[review_feedback] result.get(output, 审查Agent未返回结果。) print(f审查意见{state[review_feedback][:500]}...) # 打印前500字符 return state4.2 使用LangGraph编排工作流LangGraph允许我们以图Graph的形式定义Agent之间的交互流程。# workflows/dev_workflow.py from langgraph.graph import StateGraph, END from ..utils.state import AgentState from ..agents.planner import planning_node from ..agents.coder import coding_node from ..agents.reviewer import review_node # 假设tester也已实现 from ..agents.tester import testing_node def create_development_workflow(): 创建并返回一个开发工作流图。 # 1. 创建图并指定状态类型 workflow StateGraph(AgentState) # 2. 添加节点每个节点是一个函数接收和返回状态 workflow.add_node(planner, planning_node) workflow.add_node(coder, coding_node) workflow.add_node(reviewer, review_node) workflow.add_node(tester, testing_node) # 测试节点 # 3. 设置入口点 workflow.set_entry_point(planner) # 4. 定义边节点之间的流转逻辑 workflow.add_edge(planner, coder) workflow.add_edge(coder, reviewer) # 审查后无论是否有问题都进入测试阶段在实际中可以设置条件边 workflow.add_edge(reviewer, tester) # 5. 设置结束点 workflow.add_edge(tester, END) # 6. 编译图 return workflow.compile() # 条件边的示例高级用法 # def should_continue(state): # if 严重错误 in state.get(review_feedback, ): # return rewriter # 跳转到重写节点 # else: # return tester # 继续测试 # workflow.add_conditional_edges(reviewer, should_continue)4.3 创建主程序并运行最后我们创建主程序来驱动整个工作流。# main.py import asyncio from workflows.dev_workflow import create_development_workflow from utils.state import AgentState async def main(): # 1. 初始化工作流 app create_development_workflow() # 2. 定义初始状态用户需求 initial_state: AgentState { requirements: 开发一个Python函数用于计算斐波那契数列的第n项并为其编写单元测试。, plan: [], current_step: 0, generated_code: None, code_file_path: None, review_feedback: None, test_code: None, test_file_path: None, test_result: None, final_output: None, messages: [] } print(*50) print(启动多Agent自动化开发系统...) print(f需求{initial_state[requirements]}) print(*50) # 3. 运行工作流 try: # LangGraph的编译图可以像函数一样调用 final_state await app.ainvoke(initial_state) # 4. 输出最终结果 print(\n *50) print(工作流执行完成) print(*50) if final_state.get(code_file_path): print(f生成的代码文件{final_state[code_file_path]}) if final_state.get(review_feedback): print(f\n代码审查摘要\n{final_state[review_feedback][:1000]}...) if final_state.get(test_result): print(f\n测试结果\n{final_state[test_result]}) if final_state.get(final_output): print(f\n系统总结{final_state[final_output]}) except Exception as e: print(f工作流执行过程中出现错误{e}) if __name__ __main__: asyncio.run(main())4.4 运行与验证在项目根目录下运行主程序python main.py你将看到控制台输出类似以下内容展示了多Agent协作的完整过程 启动多Agent自动化开发系统... 需求开发一个Python函数用于计算斐波那契数列的第n项并为其编写单元测试。 阶段1任务规划 原始需求开发一个Python函数用于计算斐波那契数列的第n项并为其编写单元测试。 生成计划[1. 分析需求定义一个名为fibonacci的函数接受整数n作为参数。, 2. 实现fibonacci函数考虑递归或迭代方式并添加输入验证。, 3. 创建一个测试文件test_fibonacci.py。, 4. 在测试文件中使用pytest编写针对fibonacci函数的测试用例覆盖正常情况和边界情况。, 5. 运行pytest执行测试确保所有测试通过。] 阶段2代码生成 执行步骤1. 分析需求定义一个名为fibonacci的函数接受整数n作为参数。 进入新的AgentExecutor链... 思考我需要创建一个Python文件并写入fibonacci函数。 行动write_file 行动输入{file_path: fibonacci.py, content: def fibonacci(n):\n \\\返回斐波那契数列的第n项。\n Args:\n n (int): 非负整数。\n Returns:\n int: 第n项的值。\n \\\\n if not isinstance(n, int) or n 0:\n raise ValueError(\输入必须是非负整数\)\n if n 1:\n return n\n a, b 0, 1\n for _ in range(2, n 1):\n a, b b, a b\n return b\n} 观察成功将内容写入文件fibonacci.py 思考我现在知道最终答案了 最终答案已创建文件fibonacci.py其中包含fibonacci函数。 ... 后续阶段审查、测试依次执行5. 常见问题与排查思路在实际搭建和运行过程中你可能会遇到以下问题问题现象常见原因解决思路ModuleNotFoundError: No module named langchain1. 虚拟环境未激活。2. 依赖未正确安装。1. 确认终端前缀有(venv)使用source venv/bin/activate激活。2. 在项目根目录重新执行pip install -r requirements.txt。openai.AuthenticationErrorOpenAI API Key 未设置或错误。1. 检查.env文件是否存在且OPENAI_API_KEY值正确。2. 在代码中打印os.getenv(“OPENAI_API_KEY”)前几位确认已加载。3. 确保Key有余额和相应权限。Agent 陷入循环或输出无关内容1. Prompt指令不清晰。2. Temperature参数过高。3. 工具定义不准确。1. 优化Agent的system_prompt使其任务更明确。2. 将temperature调低如0.1。3. 检查Tool的description和参数schema是否清晰。工作流在某个节点卡住或无响应1. LLM API调用超时或限速。2. 某个Tool执行异常如文件权限不足。3. 状态State更新逻辑有误。1. 增加API调用的超时设置或添加重试逻辑。2. 在每个Tool函数内部添加更详细的异常捕获和日志。3. 使用print或logging在每个节点开始和结束时打印状态进行调试。生成的代码质量不高或不符合要求1. 给Coder Agent的上下文信息不足。2. 缺乏明确的代码规范约束。1. 在Prompt中提供更详细的约束如“必须使用类型注解”、“必须包含docstring”。2. 让Planner Agent生成更细粒度的任务描述。3. 引入Linter工具如flake8作为Reviewer Agent的检查工具之一。多Agent协作效率低下耗时过长1. 串行工作流每个步骤都等待LLM响应。2. 任务分解过细。1. 考虑对无依赖的任务使用并行节点LangGraph支持。2. 合并一些简单步骤或让一个Agent处理关联性强的多个子任务。6. 最佳实践与工程建议将多Agent系统用于实际生产环境需要超越“跑通Demo”的层面关注稳定性、安全性和可维护性。6.1 设计可观测性与日志结构化日志使用logging模块为不同Agent和Tool设置不同日志级别INFO, DEBUG, ERROR。记录每次LLM调用、Tool执行的输入输出。追踪与溯源为每个用户请求生成唯一trace_id并贯穿整个工作流。这有助于在出现问题时快速定位是哪个Agent、哪次调用出的错。状态快照定期或将关键节点的状态保存到数据库或文件便于事后分析和回滚。6.2 提升系统稳定性超时与重试对所有外部调用LLM API、数据库、网络请求设置合理的超时并实现指数退避的重试机制。熔断与降级如果某个Agent或Tool频繁失败应能暂时将其“熔断”并切换到备用方案如使用更简单的规则引擎或通知人工接管。输入验证与清理对所有来自用户或上游Agent的输入进行严格的验证和清理防止Prompt注入、代码注入等攻击。特别是在执行write_file、run_pytest这类工具时。6.3 优化Prompt工程角色扮演为每个Agent设计鲜明、具体的角色如“你是一位有10年Python经验的架构师”。少样本学习Few-Shot在Prompt中提供1-3个高质量的任务输入输出示例能显著提升Agent输出的稳定性和质量。思维链Chain-of-Thought鼓励Agent在输出最终答案前展示其推理过程如同我们使用的ReAct格式这不仅有助于调试有时也能提高答案准确性。6.4 工作流设计模式并行与聚合对于独立的任务如同时审查多个文件使用LangGraph的并行节点提高效率。条件路由利用add_conditional_edges根据上一个节点的结果动态决定下一步走向。例如审查不通过则返回给Coder修改通过则进入测试。人工审核节点在关键环节如代码合并前、生产部署前插入“人工审核”节点工作流在此暂停等待人工确认后再继续。6.5 安全与权限最小权限原则每个Agent只配备完成其任务所必需的最少工具。例如Tester Agent不需要write_file工具。沙箱环境对于执行任意代码如运行测试、安装依赖的Tool应在安全的沙箱环境如Docker容器中运行与主机隔离。敏感信息处理确保API Keys、数据库密码等不会通过LLM泄露。使用环境变量或安全的密钥管理服务。通过遵循以上实践你的多Agent自动化开发系统将从一个脆弱的实验原型进化成一个可在团队内部小范围试用的、有价值的工程辅助工具。记住当前阶段的AI Agent并非取代开发者而是作为强大的“副驾驶”处理那些定义明确、模式固定的繁琐任务从而让开发者能更专注于创造性和架构性工作。