企业官网建设流程全解析

前面几章已经分别实现了基础对话 Agent、工具调用 Agent、带记忆的对话 Agent、检索增强 Agent，以及简单任务规划 Agent。但真实业务场景通常不会只依赖一个 Agent 完成所有事情。更合理的做法是把不同能力拆给不同 Agent，再通过工作流把它们串联起来。

例如用户提出：

查询郑州今天的天气，并分析是否适合户外运动，最后生成出行建议。

这个任务至少包含四类能力：

1. 检索事实：查询天气、时间、汇率、知识库内容；
2. 分析判断：根据事实判断是否适合运动、是否需要带伞；
3. 质量审核：检查结论是否完整、安全、是否存在明显错误；
4. 最终执行：把前面所有信息整理成面向用户的自然回复。

如果把这些能力全部写进一个提示词，提示词会变得臃肿，调试也会更困难。把它们拆成多个 Agent 后，每个 Agent 的职责更清晰，也更容易单独测试和扩展。

6.1 多 Agent 拆分原则

多 Agent 不是“Agent 越多越好”，而是“能力边界越清楚越好”。设计时可以遵循以下原则：

6.2 四个 Agent 的角色设计

本章示例使用四个 Agent：

研究员负责“查事实”，推理员负责“做判断”，审核员负责“把关”，执行员负责“交付”。这种分工可以把复杂任务拆成更可控的小步骤。

6.3 完整示例代码：agent_multi.py

""" 第六章示例：多 Agent 协作与工作流编排 本章目标： 让不同 Agent 分别负责检索、推理、审核和执行，形成更复杂的自动化流程。 运行方式： 1. 确保同一目录下已有 agent_with_rag.py 2. 在项目根目录创建 .env 文件，写入 OPENAI_API_KEY 3. 安装依赖： pip install langchain langchain-openai faiss-cpu python-dotenv 4. 运行： python agent_multi.py """ import os import json from datetime import datetime from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage # ---------------------------- # 1. 加载环境变量 # ---------------------------- load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o") MODEL_BASE_URL = os.getenv("MODEL_BASE_URL", "").strip() if not OPENAI_API_KEY: raise RuntimeError( "未找到 OPENAI_API_KEY，请在项目根目录创建 .env 文件并写入 API Key。" ) # ---------------------------- # 2. 初始化基础模型 # ---------------------------- llm_kwargs = { "model": MODEL_NAME, "temperature": 0.2, "api_key": OPENAI_API_KEY, } if MODEL_BASE_URL: llm_kwargs["base_url"] = MODEL_BASE_URL base_llm = ChatOpenAI(**llm_kwargs) # ---------------------------- # 3. 从第五章导入检索增强能力 # ---------------------------- # 这里复用第五章已经写好的知识库检索工具， # 避免重复实现向量索引、文本切分和 FAISS 加载逻辑。 try: from agent_with_rag import search_knowledge_base except ImportError: # 如果 agent_with_rag.py 不存在，提供一个降级版本 @tool def search_knowledge_base(query: str) -> str: """在本地知识库中检索相关文档片段。""" return "当前未找到可用的知识库文档，请检查 knowledge_base 目录。" # ---------------------------- # 4. 通用工具：天气、时间、汇率 # ---------------------------- @tool def get_weather(city: str) -> str: """查询指定城市的天气信息。 参数： - city: 城市名称，例如“郑州”“北京”“上海” """ if not city or not city.strip(): return "缺少城市名称，无法查询天气。" city = city.strip() weather_data = { "郑州": "晴，22°C，微风", "北京": "多云，18°C，北风2级", "上海": "小雨，20°C，东南风3级", "广州": "阴，26°C，湿度较高", "深圳": "晴，28°C，适合出行", } if city in weather_data: return f"{city}天气：{weather_data[city]}" return f"未找到“{city}”的天气数据，请确认城市名称。" @tool def get_current_time(dummy: str = "") -> str: """获取当前日期和时间。""" now = datetime.now() return now.strftime("%Y年%m月%d日 %H:%M:%S") @tool def convert_currency(amount: float, from_currency: str, to_currency: str) -> str: """进行简单的货币换算。""" if amount <= 0: return "换算金额必须大于 0。" from_currency = from_currency.upper() to_currency = to_currency.upper() rates = { "USD_CNY": 7.2, "CNY_USD": 1 / 7.2, "EUR_CNY": 7.8, "CNY_EUR": 1 / 7.8, } key = f"{from_currency}_{to_currency}" if key not in rates: return f"暂不支持 {from_currency} 到 {to_currency} 的换算。" result = round(amount * rates[key], 2) return f"{amount} {from_currency} ≈ {result} {to_currency}" COMMON_TOOLS = [get_weather, get_current_time, convert_currency, search_knowledge_base] # ---------------------------- # 5. 创建研究员 Agent：负责检索信息 # ---------------------------- def create_researcher_agent() -> AgentExecutor: """研究员 Agent：负责检索天气、时间、汇率和本地知识库信息。 它不直接给最终建议，只负责把事实信息查清楚。 """ prompt = ChatPromptTemplate.from_messages( [ ( "system", "你是一个研究员 Agent，负责检索和整理事实信息。" "你可以调用工具查询天气、时间、汇率，也可以检索本地知识库。" "你的任务是把可靠信息整理出来，不要过度发挥或编造未查到的内容。" "如果某些信息无法查询到，请明确说明缺失部分。" ), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ] ) agent = create_tool_calling_agent(base_llm, COMMON_TOOLS, prompt) return AgentExecutor( agent=agent, tools=COMMON_TOOLS, verbose=False, max_iterations=6, handle_parsing_errors=True, ) # ---------------------------- # 6. 创建推理员 Agent：负责分析和判断 # ---------------------------- def create_reasoner_agent() -> AgentExecutor: """推理员 Agent：基于已有信息进行分析和判断。 它通常接收研究员的输出，然后回答“这意味着什么”“是否适合”“有什么风险”等问题。 """ prompt = ChatPromptTemplate.from_messages( [ ( "system", "你是一个推理员 Agent，负责基于已有信息进行分析和判断。" "你需要解释信息背后的含义，给出合理推断，但不要编造未提供的事实。" "如果输入信息不足，请指出还需要补充哪些信息。" "回答要清晰、有条理，适合继续交给审核员检查。" ), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ] ) # 推理员主要做分析，这里不额外给它工具， # 如果需要也可以传入 COMMON_TOOLS，让它自行补查。 agent = create_tool_calling_agent(base_llm, [], prompt) return AgentExecutor( agent=agent, tools=[], verbose=False, max_iterations=6, handle_parsing_errors=True, ) # ---------------------------- # 7. 创建审核员 Agent：负责质量检查 # ---------------------------- def create_reviewer_agent() -> AgentExecutor: """审核员 Agent：检查前序输出是否完整、准确、安全。 它会检查： 1. 是否回答了用户原始问题； 2. 是否有明显事实错误； 3. 是否有未说明的假设； 4. 是否存在不安全或不合适的建议。 """ prompt = ChatPromptTemplate.from_messages( [ ( "system", "你是一个审核员 Agent，负责检查前序输出的质量。" "请从以下角度审查：" "1. 是否回答了用户原始问题；" "2. 是否存在明显事实错误或自相矛盾；" "3. 是否包含未说明的重要假设；" "4. 是否存在不安全、不合适或过度确定的表述；" "5. 是否需要补充更多信息。" "如果发现严重问题，请指出并给出修改建议。" "如果没有明显问题，请给出简短通过结论。" ), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ] ) agent = create_tool_calling_agent(base_llm, [], prompt) return AgentExecutor( agent=agent, tools=[], verbose=False, max_iterations=6, handle_parsing_errors=True, ) # ---------------------------- # 8. 创建执行员 Agent：负责生成最终回复 # ---------------------------- def create_executor_agent() -> AgentExecutor: """执行员 Agent：综合前面所有信息，生成面向用户的最终回答。 它会把检索结果、推理结论、审核意见整合成自然、清晰、可执行的回复。 """ prompt = ChatPromptTemplate.from_messages( [ ( "system", "你是一个执行员 Agent，负责生成最终面向用户的回答。" "你需要综合研究员、推理员、审核员的信息，输出清晰、自然、可执行的结论。" "如果前面存在不确定信息，请不要假装确定，而要如实说明。" "回答要简洁友好，避免堆砌内部流程细节。" ), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ] ) agent = create_tool_calling_agent(base_llm, [], prompt) return AgentExecutor( agent=agent, tools=[], verbose=False, max_iterations=6, handle_parsing_errors=True, ) # ---------------------------- # 9. 工作流编排器 # ---------------------------- class WorkflowOrchestrator: """工作流编排器：协调多个 Agent 完成复杂任务。 支持两种模式： 1. 链式工作流：按顺序执行，前一个 Agent 的输出作为后一个的输入 2. 条件工作流：根据任务关键词选择不同执行路径 """ def __init__(self, agents: dict): """初始化编排器。 参数： - agents: Agent 字典，例如 { "researcher": researcher_agent, "reasoner": reasoner_agent, "reviewer": reviewer_agent, "executor": executor_agent, } """ self.agents = agents self.workflow_logs = [] def log_step(self, step_name: str, agent_name: str, input_text: str, output_text: str): """记录工作流中的每一步执行。""" self.workflow_logs.append({ "step": step_name, "agent": agent_name, "input": input_text[:200], "output": output_text[:500], "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }) print(f" [{step_name}] {agent_name} 已完成") def run_chain_workflow(self, task: str, workflow: list) -> str: """运行链式工作流。 参数： - task: 用户的原始任务 - workflow: Agent 名称列表，按执行顺序排列 返回： - 最终执行结果 """ print(f"\n{'='*60}") print(f"开始链式工作流：{task}") print(f"执行顺序：{' -> '.join(workflow)}") print(f"{'='*60}") self.workflow_logs = [] current_input = task for agent_name in workflow: if agent_name not in self.agents: return f"错误：未找到 Agent '{agent_name}'" agent = self.agents[agent_name] print(f"\n 执行：{agent_name}") print(f" 输入：{current_input[:100]}...") try: result = agent.invoke({"input": current_input}) output = result.get("output", "无输出") except Exception as e: output = f"执行失败：{str(e)}" self.log_step( step_name=f"步骤_{workflow.index(agent_name) + 1}", agent_name=agent_name, input_text=current_input, output_text=output, ) # 下一个 Agent 的输入 = 当前 Agent 的输出 current_input = output print(f"\n{'='*60}") print("工作流执行完成！") print(f"{'='*60}") return current_input def run_conditional_workflow( self, task: str, conditions: list, default_workflow: list ) -> str: """运行条件工作流。 参数： - task: 用户的原始任务 - conditions: 条件规则列表，例如 [ { "keywords": ["天气", "气温", "降雨"], "workflow": ["researcher", "reasoner", "executor"] } ] - default_workflow: 默认执行流程 返回： - 最终执行结果 """ print(f"\n{'='*60}") print(f"开始条件工作流：{task}") print(f"{'='*60}") self.workflow_logs = [] selected_workflow = default_workflow for condition in conditions: keywords = condition.get("keywords", []) workflow = condition.get("workflow", []) if any(kw in task for kw in keywords): selected_workflow = workflow print(f" 匹配条件：关键词 {keywords}") print(f" 选择工作流：{' -> '.join(workflow)}") break if selected_workflow == default_workflow: print(f" 未匹配特定条件，使用默认工作流：{' -> '.join(default_workflow)}") return self.run_chain_workflow(task, selected_workflow) def get_workflow_logs(self) -> list: """获取工作流执行日志。""" return self.workflow_logs def save_workflow_logs(self, filename: str = "workflow_logs.json"): """将工作流日志保存到文件。""" with open(filename, "w", encoding="utf-8") as f: json.dump(self.workflow_logs, f, ensure_ascii=False, indent=2) print(f"\n工作流日志已保存到：{filename}") # ---------------------------- # 10. 主程序入口 # ---------------------------- def main(): """演示多 Agent 协作与工作流编排。""" print("=" * 60) print("第六章示例：多 Agent 协作与工作流编排") print("=" * 60) # 1. 创建各个 Agent researcher_agent = create_researcher_agent() reasoner_agent = create_reasoner_agent() reviewer_agent = create_reviewer_agent() executor_agent = create_executor_agent() # 2. 初始化工作流编排器 orchestrator = WorkflowOrchestrator({ "researcher": researcher_agent, "reasoner": reasoner_agent, "reviewer": reviewer_agent, "executor": executor_agent, }) # 3. 演示场景1：链式工作流 print("\n" + "=" * 60) print("场景1：链式工作流 - 完整的信息处理流程") print("=" * 60) task1 = "查询郑州今天的天气，并分析是否适合户外运动，最后生成建议" workflow1 = [ "researcher", # 先检索天气信息 "reasoner", # 再分析是否适合运动 "reviewer", # 然后审核分析结果 "executor", # 最后执行并生成建议 ] result1 = orchestrator.run_chain_workflow(task1, workflow1) print(f"\n最终结果：\n{result1}") # 4. 演示场景2：条件工作流 print("\n" + "=" * 60) print("场景2：条件工作流 - 根据任务类型选择不同流程") print("=" * 60) conditions = [ { "keywords": ["天气", "气温", "降雨"], "workflow": ["researcher", "reasoner", "executor"], }, { "keywords": ["代码", "编程", "算法"], "workflow": ["reasoner", "reviewer", "executor"], }, ] default_workflow = ["researcher", "reasoner", "executor"] task2 = "今天郑州下雨吗？需要带伞吗？" result2 = orchestrator.run_conditional_workflow(task2, conditions, default_workflow) print(f"\n最终结果：\n{result2}") # 5. 演示场景3：交互式对话 print("\n" + "=" * 60) print("场景3：交互式多 Agent 对话") print("=" * 60) print("说明：") print("- 输入 'chain <任务>' 使用链式工作流") print("- 输入 'auto <任务>' 使用条件工作流") print("- 输入 'q' 退出") print("=" * 60) while True: try: user_input = input("\n你：").strip() except (EOFError, KeyboardInterrupt): print("\n再见！") break if not user_input: continue if user_input.lower() in ("q", "quit", "exit", "退出"): orchestrator.save_workflow_logs() print("再见！") break if user_input.startswith("chain "): task = user_input[6:].strip() workflow = ["researcher", "reasoner", "reviewer", "executor"] result = orchestrator.run_chain_workflow(task, workflow) print(f"\nAI：{result}") elif user_input.startswith("auto "): task = user_input[5:].strip() result = orchestrator.run_conditional_workflow(task, conditions, default_workflow) print(f"\nAI：{result}") else: print("请使用 'chain <任务>' 或 'auto <任务>' 格式输入") if __name__ == "__main__": main()

6.4 工作流执行流程示意

链式工作流的执行顺序如下：

1用户任务 2 ↓ 3研究员 Agent：查询天气、时间、汇率、知识库 4 ↓ 5推理员 Agent：基于事实进行分析 6 ↓ 7审核员 Agent：检查结论是否完整、安全 8 ↓ 9执行员 Agent：生成最终回复

条件工作流会根据任务关键词选择不同路径。例如：

1任务包含“天气”“气温”“降雨” 2 → researcher → reasoner → executor 3 4任务包含“代码”“编程”“算法” 5 → reasoner → reviewer → executor 6 7其他任务 8 → researcher → reasoner → executor（默认流程）

条件工作流适合处理“不同问题需要不同能力组合”的场景。比如天气类任务需要先查事实，代码类任务可能更依赖推理和审核，而通用任务则走默认流程。

6.5 运行与验证步骤

运行前请确认项目目录结构大致如下：

1project/ 2├── .env 3├── agent_with_rag.py 4├── agent_multi.py 5└── verify_multi_agent.py

运行步骤：

1. 在项目根目录创建.env文件，写入：

1OPENAI_API_KEY=你的API密钥 2MODEL_NAME=gpt-4o 3MODEL_BASE_URL=

如果使用的是兼容 OpenAI 接口的模型，可以填写MODEL_BASE_URL；否则留空即可。

1. 安装依赖：

1pip install langchain langchain-openai faiss-cpu python-dotenv

1. 运行主程序：

1python agent_multi.py

1. 交互式输入示例：

1chain 查询郑州今天的天气，并分析是否适合户外运动 2auto 今天郑州下雨吗？需要带伞吗？ 3q

运行结束后，工作流日志会保存为workflow_logs.json，方便后续调试。

6.6 验证脚本：verify_multi_agent.py

这个脚本用于快速验证多 Agent 是否按预期工作，而不需要进入交互式对话。

1""" 2多 Agent 工作流验证脚本 3 4用途： 5自动运行几个典型任务，检查研究员、推理员、审核员、执行员是否能正常协作。 6 7运行方式： 8python verify_multi_agent.py 9""" 10 11from agent_multi import ( 12 create_researcher_agent, 13 create_reasoner_agent, 14 create_reviewer_agent, 15 create_executor_agent, 16 WorkflowOrchestrator, 17) 18 19 20def verify_chain_workflow(): 21 """验证链式工作流。""" 22 print("\n" + "=" * 60) 23 print("验证1：链式工作流") 24 print("=" * 60) 25 26 orchestrator = WorkflowOrchestrator({ 27 "researcher": create_researcher_agent(), 28 "reasoner": create_reasoner_agent(), 29 "reviewer": create_reviewer_agent(), 30 "executor": create_executor_agent(), 31 }) 32 33 task = "查询郑州今天的天气，并分析是否适合户外运动，最后生成建议" 34 result = orchestrator.run_chain_workflow( 35 task, 36 ["researcher", "reasoner", "reviewer", "executor"], 37 ) 38 39 print(f"\n最终回复：\n{result}") 40 41 # 简单检查：最终结果不应为空 42 assert result and len(result.strip()) > 10, "链式工作流未返回有效结果" 43 print("\n链式工作流验证通过。") 44 45 46def verify_conditional_workflow(): 47 """验证条件工作流。""" 48 print("\n" + "=" * 60) 49 print("验证2：条件工作流") 50 print("=" * 60) 51 52 orchestrator = WorkflowOrchestrator({ 53 "researcher": create_researcher_agent(), 54 "reasoner": create_reasoner_agent(), 55 "reviewer": create_reviewer_agent(), 56 "executor": create_executor_agent(), 57 }) 58 59 conditions = [ 60 { 61 "keywords": ["天气", "气温", "降雨"], 62 "workflow": ["researcher", "reasoner", "executor"], 63 }, 64 { 65 "keywords": ["代码", "编程", "算法"], 66 "workflow": ["reasoner", "reviewer", "executor"], 67 }, 68 ] 69 70 default_workflow = ["researcher", "reasoner", "executor"] 71 72 task = "今天郑州下雨吗？需要带伞吗？" 73 result = orchestrator.run_conditional_workflow(task, conditions, default_workflow) 74 75 print(f"\n最终回复：\n{result}") 76 77 assert result and len(result.strip()) > 10, "条件工作流未返回有效结果" 78 print("\n条件工作流验证通过。") 79 80 81def verify_single_agent(): 82 """验证单个 Agent 是否可以独立运行。""" 83 print("\n" + "=" * 60) 84 print("验证3：单个 Agent 独立运行") 85 print("=" * 60) 86 87 researcher = create_researcher_agent() 88 result = researcher.invoke({"input": "查询郑州今天的天气"}) 89 output = result.get("output", "") 90 91 print(f"研究员输出：\n{output}") 92 93 assert "郑州" in output or "天气" in output, "研究员未返回预期天气信息" 94 print("\n单个 Agent 验证通过。") 95 96 97if __name__ == "__main__": 98 verify_single_agent() 99 verify_chain_workflow() 100 verify_conditional_workflow() 101 102 print("\n" + "=" * 60) 103 print("所有验证完成。") 104 print("=" * 60)

运行方式：

1python verify_multi_agent.py

如果某个 Agent 调用失败，脚本会在对应断言处报错，方便定位问题。

6.7 本章小结

这一章把前面几章的能力组合起来，从“单个 Agent”推进到“多 Agent 协作”。研究员负责查事实，推理员负责做判断，审核员负责把关，执行员负责输出最终回复；工作流编排器则负责把它们按链式或条件式流程串联起来。

多 Agent 的价值不在于堆数量，而在于让每个环节的职责更清晰。调试时，你可以单独检查研究员是否查到了正确信息，推理员是否做出了合理判断，审核员是否发现了漏洞，执行员是否把结论表达清楚。这样比把所有逻辑塞进一个提示词更容易维护和扩展。