{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 20 多智能体架构\n", "\n", "## 学习目标\n", "1. 理解多智能体系统的典型架构（如监督者、协作、竞争）\n", "2. 掌握使用LangGraph设计多节点多智能体工作流\n", "3. 能够实现多个智能体节点之间的消息传递与协作" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. 为什么需要多智能体\n", "\n", "在之前的课程中，我们构建的智能体都是**单智能体**：一个智能体完成所有任务。\n", "\n", "但真实场景中，很多任务需要**多个智能体协作**才能完成：\n", "\n", "- **复杂任务分解**：一个任务太大，需要分给多个专业智能体\n", "- **专业分工**：不同智能体有不同的专业知识（如代码专家、写作专家、数据分析专家）\n", "- **协作决策**：需要多个智能体共同讨论才能做出最佳决策\n", "- **竞争博弈**：多个智能体竞争完成同一任务\n", "\n", "简单来说：\n", "\n", "- **单智能体**：一个人做所有事\n", "- **多智能体**：一个团队分工合作" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. 多智能体的典型架构模式\n", "\n", "多智能体系统有几种经典的架构模式：\n", "\n", "### 2.1 监督者模式（Supervisor Pattern）\n", "\n", "一个监督者智能体负责协调多个工作智能体：\n", "\n", "```\n", " 监督者\n", " │\n", " ┌──────┼──────┬──────┐\n", " ▼ ▼ ▼ ▼\n", " 智能体A 智能体B 智能体C 智能体D\n", "```\n", "\n", "**适用场景**：任务需要多种专业能力，监督者分配任务并汇总结果。\n", "\n", "### 2.2 协作模式（Collaborative Pattern）\n", "\n", "多个智能体平等协作，共同完成任务：\n", "\n", "```\n", " 智能体A ←→ 智能体B\n", " ↑ ↑\n", " │ │\n", " 智能体C ←→ 智能体D\n", "```\n", "\n", "**适用场景**：任务需要多个智能体之间频繁交流和协同。\n", "\n", "### 2.3 竞争模式（Competitive Pattern）\n", "\n", "多个智能体竞争完成同一任务，胜者获得奖励：\n", "\n", "```\n", " 智能体A ─┐\n", " ├──→ 比较器 → 最佳结果\n", " 智能体B ─┘\n", "```\n", "\n", "**适用场景**：需要从多个方案中选择最优解。\n", "\n", "### 2.4 流水线模式（Pipeline Pattern）\n", "\n", "多个智能体按顺序处理任务，像流水线一样：\n", "\n", "```\n", " 智能体A → 智能体B → 智能体C → 智能体D\n", "```\n", "\n", "**适用场景**：任务可以分解为多个顺序步骤。\n", "\n", "### 2.5 辩论模式（Debate Pattern）\n", "\n", "多个智能体就某个问题展开辩论，最终达成共识：\n", "\n", "```\n", " 正方智能体 ←→ 反方智能体\n", " │ │\n", " └────┬──────┘\n", " ▼\n", " 裁判智能体\n", "```\n", "\n", "**适用场景**：需要深入分析某个问题的正反两面。" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. 第一个例子：简单的协作智能体\n", "\n", "我们用一个简单的例子来演示多智能体协作。\n", "\n", "场景：\n", "- 用户提出一个问题\n", "- 研究智能体负责收集信息\n", "- 写作智能体负责整理成报告\n", "- 监督者智能体负责协调整个流程" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing_extensions import TypedDict\n", "from langgraph.graph import StateGraph, START, END\n", "from openai import OpenAI\n", "import os\n", "from dotenv import load_dotenv\n", "\n", "load_dotenv()\n", "\n", "client = OpenAI(\n", " base_url=os.getenv(\"OPENAI_BASE_URL\"),\n", " api_key=os.getenv(\"OPENAI_API_KEY\")\n", ")\n", "\n", "def call_llm(system_prompt: str, user_prompt: str) -> str:\n", " response = client.chat.completions.create(\n", " model=\"qwen3.6-35b-A3b\",\n", " messages=[\n", " {\"role\": \"system\", \"content\": system_prompt},\n", " {\"role\": \"user\", \"content\": user_prompt}\n", " ],\n", " temperature=0.7,\n", " max_tokens=500\n", " )\n", " return response.choices[0].message.content\n", "\n", "class AgentState(TypedDict):\n", " user_question: str\n", " research_info: str\n", " report: str\n", " current_agent: str\n", "\n", "def supervisor(state: AgentState):\n", " print(f'监督者：收到用户问题：{state[\"user_question\"]}')\n", " if not state['research_info']:\n", " print('监督者：分配给研究智能体')\n", " return {'current_agent': 'research'}\n", " elif not state['report']:\n", " print('监督者：分配给写作智能体')\n", " return {'current_agent': 'writer'}\n", " else:\n", " print('监督者：任务完成')\n", " return {'current_agent': 'finish'}\n", "\n", "def research_agent(state: AgentState):\n", " print(f'研究智能体：正在研究 \"{state[\"user_question\"]}\"')\n", " system_prompt = '你是一个专业的研究助手，请针对用户问题提供详细的研究信息和分析。'\n", " user_prompt = f'请研究并总结关于\"{state[\"user_question\"]}\"的最新信息，包括主要趋势、关键技术和未来展望。'\n", " research_info = call_llm(system_prompt, user_prompt)\n", " print(f'研究智能体：完成研究，收集到信息')\n", " return {'research_info': research_info}\n", "\n", "def writer_agent(state: AgentState):\n", " print(f'写作智能体：正在撰写报告')\n", " system_prompt = '你是一个专业的报告撰写者，请根据提供的研究信息撰写结构化的报告。'\n", " user_prompt = f'请根据以下研究信息撰写一份完整的报告：\\n\\n研究信息：{state[\"research_info\"]}'\n", " report = call_llm(system_prompt, user_prompt)\n", " print(f'写作智能体：报告完成')\n", " return {'report': report}\n", "\n", "def route(state: AgentState):\n", " if state['current_agent'] == 'research':\n", " return 'research'\n", " elif state['current_agent'] == 'writer':\n", " return 'writer'\n", " elif state['current_agent'] == 'finish':\n", " return END\n", " return 'supervisor'\n", "\n", "builder = StateGraph(AgentState)\n", "builder.add_node('supervisor', supervisor)\n", "builder.add_node('research', research_agent)\n", "builder.add_node('writer', writer_agent)\n", "\n", "builder.add_edge(START, 'supervisor')\n", "builder.add_edge('research', 'supervisor')\n", "builder.add_edge('writer', 'supervisor')\n", "builder.add_conditional_edges(\n", " 'supervisor',\n", " route,\n", " {\n", " 'research': 'research',\n", " 'writer': 'writer',\n", " END: END\n", " }\n", ")\n", "\n", "graph = builder.compile()\n", "\n", "result = graph.invoke({\n", " 'user_question': '人工智能的发展趋势',\n", " 'research_info': '',\n", " 'report': '',\n", " 'current_agent': ''\n", "})\n", "\n", "print()\n", "print('最终报告：')\n", "print(result['report'])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "1. **环境配置**：\n", " - 加载 `.env` 文件中的 API 密钥\n", " - 创建 OpenAI 客户端\n", " - 定义 `call_llm` 函数，封装大模型调用逻辑\n", "\n", "2. **AgentState**：定义状态包含四个字段：\n", " - `user_question`：用户的问题\n", " - `research_info`：研究智能体收集的信息\n", " - `report`：写作智能体生成的报告\n", " - `current_agent`：当前应该执行的智能体\n", "\n", "3. **supervisor**：监督者智能体\n", " - 检查当前状态，决定下一步分配给哪个智能体\n", " - 如果没有研究信息，分配给研究智能体\n", " - 如果有研究信息但没有报告，分配给写作智能体\n", " - 如果报告已完成，标记任务结束\n", "\n", "4. **research_agent**：研究智能体（使用真实大模型）\n", " - 构建系统提示词和用户提示词\n", " - 调用大模型获取研究结果\n", " - 将结果保存到 `research_info`\n", "\n", "5. **writer_agent**：写作智能体（使用真实大模型）\n", " - 构建系统提示词和用户提示词\n", " - 调用大模型根据研究信息生成报告\n", " - 将结果保存到 `report`\n", "\n", "6. **route**：路由函数\n", " - 根据 `current_agent` 字段决定下一步执行哪个节点\n", "\n", "7. **图结构**：\n", " - START -> supervisor\n", " - supervisor 根据条件边分配给 research 或 writer\n", " - research 和 writer 执行完后都回到 supervisor\n", " - supervisor 判断完成后走到 END\n", "\n", "流程图：\n", "\n", "```\n", "START -> supervisor\n", " │\n", " ┌──────┴──────┐\n", " ▼ ▼\n", "research writer\n", " │ │\n", " └──────┬──────┘\n", " ▼\n", " supervisor -> END\n", "```" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. 流水线模式：多智能体顺序处理\n", "\n", "流水线模式是最直观的多智能体架构，智能体按顺序处理任务。\n", "\n", "场景：\n", "- 用户输入一段原始文本\n", "- 翻译智能体将其翻译成英文\n", "- 摘要智能体生成摘要\n", "- 情感分析智能体分析情感倾向" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing_extensions import TypedDict\n", "from langgraph.graph import StateGraph, START, END\n", "\n", "class PipelineState(TypedDict):\n", " original_text: str\n", " translated_text: str\n", " summary: str\n", " sentiment: str\n", "\n", "def translate_agent(state: PipelineState):\n", " print(f'翻译智能体：正在翻译')\n", " text = state['original_text']\n", " translated = f'[Translated] {text}'\n", " print(f'翻译智能体：{translated}')\n", " return {'translated_text': translated}\n", "\n", "def summarize_agent(state: PipelineState):\n", " print(f'摘要智能体：正在生成摘要')\n", " text = state['translated_text']\n", " summary = f'[Summary] {text[:20]}...'\n", " print(f'摘要智能体：{summary}')\n", " return {'summary': summary}\n", "\n", "def sentiment_agent(state: PipelineState):\n", " print(f'情感分析智能体：正在分析情感')\n", " text = state['translated_text']\n", " if '好' in text or '高兴' in text:\n", " sentiment = '积极'\n", " elif '坏' in text or '难过' in text:\n", " sentiment = '消极'\n", " else:\n", " sentiment = '中性'\n", " print(f'情感分析智能体：情感倾向为 {sentiment}')\n", " return {'sentiment': sentiment}\n", "\n", "builder = StateGraph(PipelineState)\n", "builder.add_node('translate', translate_agent)\n", "builder.add_node('summarize', summarize_agent)\n", "builder.add_node('sentiment', sentiment_agent)\n", "\n", "builder.add_edge(START, 'translate')\n", "builder.add_edge('translate', 'summarize')\n", "builder.add_edge('summarize', 'sentiment')\n", "builder.add_edge('sentiment', END)\n", "\n", "graph = builder.compile()\n", "\n", "result = graph.invoke({\n", " 'original_text': '今天天气很好，我很高兴。',\n", " 'translated_text': '',\n", " 'summary': '',\n", " 'sentiment': ''\n", "})\n", "\n", "print()\n", "print('处理结果：')\n", "print(f'原文：{result[\"original_text\"]}')\n", "print(f'翻译：{result[\"translated_text\"]}')\n", "print(f'摘要：{result[\"summary\"]}')\n", "print(f'情感：{result[\"sentiment\"]}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "1. **PipelineState**：定义流水线的状态，包含四个字段，分别对应每个智能体的输入/输出。\n", "\n", "2. **translate_agent**：翻译智能体\n", " - 接收原始文本\n", " - 输出翻译结果\n", "\n", "3. **summarize_agent**：摘要智能体\n", " - 接收翻译后的文本\n", " - 输出摘要\n", "\n", "4. **sentiment_agent**：情感分析智能体\n", " - 接收翻译后的文本\n", " - 分析情感倾向\n", "\n", "5. **图结构**：简单的顺序流程\n", " - START -> translate -> summarize -> sentiment -> END\n", "\n", "流程图：\n", "\n", "```\n", "START -> translate -> summarize -> sentiment -> END\n", "```\n", "\n", "这种模式的优点：\n", "- 结构简单，易于理解\n", "- 每个智能体只关注自己的任务\n", "- 便于扩展，可以随时添加新的智能体" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. 竞争模式：多个智能体竞争\n", "\n", "竞争模式中，多个智能体同时处理同一个任务，然后比较结果选择最优解。\n", "\n", "场景：\n", "- 用户提出一个问题\n", "- 多个回答智能体给出不同的答案\n", "- 评估智能体选择最佳答案" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing_extensions import TypedDict\n", "from langgraph.graph import StateGraph, START, END\n", "\n", "class CompetitionState(TypedDict):\n", " question: str\n", " answers: list\n", " best_answer: str\n", "\n", "def answer_agent_a(state: CompetitionState):\n", " print(f'回答智能体A：正在回答问题')\n", " answer = f'A的回答：关于\"{state[\"question\"]}\"，这是一个详细的回答。'\n", " print(f'回答智能体A：{answer}')\n", " return {'answers': state['answers'] + [{'agent': 'A', 'content': answer, 'score': 85}]}\n", "\n", "def answer_agent_b(state: CompetitionState):\n", " print(f'回答智能体B：正在回答问题')\n", " answer = f'B的回答：对于\"{state[\"question\"]}\"，我的见解是...'\n", " print(f'回答智能体B：{answer}')\n", " return {'answers': state['answers'] + [{'agent': 'B', 'content': answer, 'score': 92}]}\n", "\n", "def answer_agent_c(state: CompetitionState):\n", " print(f'回答智能体C：正在回答问题')\n", " answer = f'C的回答：\"{state[\"question\"]}\"的答案如下...'\n", " print(f'回答智能体C：{answer}')\n", " return {'answers': state['answers'] + [{'agent': 'C', 'content': answer, 'score': 88}]}\n", "\n", "def evaluator_agent(state: CompetitionState):\n", " print(f'评估智能体：正在评估所有回答')\n", " if not state['answers']:\n", " return {'best_answer': '没有回答'}\n", "\n", " best = max(state['answers'], key=lambda x: x['score'])\n", " print(f'评估智能体：最佳回答来自智能体{best[\"agent\"]}，得分{best[\"score\"]}')\n", " return {'best_answer': best['content']}\n", "\n", "builder = StateGraph(CompetitionState)\n", "builder.add_node('answer_a', answer_agent_a)\n", "builder.add_node('answer_b', answer_agent_b)\n", "builder.add_node('answer_c', answer_agent_c)\n", "builder.add_node('evaluator', evaluator_agent)\n", "\n", "builder.add_edge(START, 'answer_a')\n", "builder.add_edge('answer_a', 'answer_b')\n", "builder.add_edge('answer_b', 'answer_c')\n", "builder.add_edge('answer_c', 'evaluator')\n", "builder.add_edge('evaluator', END)\n", "\n", "graph = builder.compile()\n", "\n", "result = graph.invoke({\n", " 'question': '什么是人工智能？',\n", " 'answers': [],\n", " 'best_answer': ''\n", "})\n", "\n", "print()\n", "print('最佳回答：')\n", "print(result['best_answer'])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "1. **CompetitionState**：定义竞争状态，包含：\n", " - `question`：用户问题\n", " - `answers`：所有智能体的回答列表\n", " - `best_answer`：评估后的最佳回答\n", "\n", "2. **answer_agent_a/b/c**：三个回答智能体\n", " - 每个智能体给出自己的回答\n", " - 回答包含智能体标识、内容和得分\n", " - 将回答追加到 `answers` 列表\n", "\n", "3. **evaluator_agent**：评估智能体\n", " - 比较所有回答的得分\n", " - 选择得分最高的回答作为最佳答案\n", "\n", "4. **图结构**：顺序执行三个回答智能体，然后评估\n", " - START -> answer_a -> answer_b -> answer_c -> evaluator -> END\n", "\n", "流程图：\n", "\n", "```\n", "START -> answer_a -> answer_b -> answer_c -> evaluator -> END\n", "```\n", "\n", "这种模式的优点：\n", "- 可以从多个角度解决问题\n", "- 通过竞争提高回答质量\n", "- 易于扩展，随时可以添加新的智能体" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 6. 辩论模式：智能体之间的辩论\n", "\n", "辩论模式中，多个智能体就某个问题展开辩论，最终达成共识或由裁判做出裁决。\n", "\n", "场景：\n", "- 用户提出一个有争议的问题\n", "- 正方智能体支持某个观点\n", "- 反方智能体反对该观点\n", "- 裁判智能体总结辩论结果" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing_extensions import TypedDict\n", "from langgraph.graph import StateGraph, START, END\n", "\n", "class DebateState(TypedDict):\n", " question: str\n", " debate_history: list\n", " round: int\n", " conclusion: str\n", "\n", "def pro_agent(state: DebateState):\n", " print(f'正方智能体（第{state[\"round\"]}轮）：')\n", " arguments = [\n", " f'支持\"{state[\"question\"]}\"的理由：它可以提高效率。',\n", " f'进一步论证：从长远来看，这是必然趋势。'\n", " ]\n", " arg = arguments[state['round'] - 1]\n", " print(f' {arg}')\n", " return {\n", " 'debate_history': state['debate_history'] + [{'side': '正方', 'argument': arg}],\n", " 'round': state['round']\n", " }\n", "\n", "def con_agent(state: DebateState):\n", " print(f'反方智能体（第{state[\"round\"]}轮）：')\n", " arguments = [\n", " f'反对\"{state[\"question\"]}\"的理由：它可能带来风险。',\n", " f'进一步反驳：我们需要更谨慎地对待。'\n", " ]\n", " arg = arguments[state['round'] - 1]\n", " print(f' {arg}')\n", " new_round = state['round'] + 1\n", " return {\n", " 'debate_history': state['debate_history'] + [{'side': '反方', 'argument': arg}],\n", " 'round': new_round\n", " }\n", "\n", "def judge_agent(state: DebateState):\n", " print(f'裁判智能体：总结辩论')\n", " conclusion = f'关于\"{state[\"question\"]}\"的辩论总结：\\n'\n", " for entry in state['debate_history']:\n", " conclusion += f'- {entry[\"side\"]}：{entry[\"argument\"]}\\n'\n", " conclusion += '\\n结论：双方观点都有道理，需要权衡利弊。'\n", " print(f'裁判智能体：{conclusion}')\n", " return {'conclusion': conclusion}\n", "\n", "def after_con_route(state: DebateState):\n", " if state['round'] <= 2:\n", " return 'pro'\n", " return 'judge'\n", "\n", "builder = StateGraph(DebateState)\n", "builder.add_node('pro', pro_agent)\n", "builder.add_node('con', con_agent)\n", "builder.add_node('judge', judge_agent)\n", "\n", "builder.add_edge(START, 'pro')\n", "builder.add_edge('pro', 'con')\n", "builder.add_conditional_edges(\n", " 'con',\n", " after_con_route,\n", " {\n", " 'pro': 'pro',\n", " 'judge': 'judge'\n", " }\n", ")\n", "builder.add_edge('judge', END)\n", "\n", "graph = builder.compile()\n", "\n", "result = graph.invoke({\n", " 'question': '人工智能是否应该取代人类工作？',\n", " 'debate_history': [],\n", " 'round': 1,\n", " 'conclusion': ''\n", "})\n", "\n", "print()\n", "print('辩论总结：')\n", "print(result['conclusion'])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "1. **DebateState**：定义辩论状态，包含：\n", " - `question`：辩论的问题\n", " - `debate_history`：辩论历史记录\n", " - `round`：当前辩论轮次\n", " - `conclusion`：最终结论\n", "\n", "2. **pro_agent**：正方智能体\n", " - 根据当前轮次给出支持观点\n", " - 将论点添加到辩论历史\n", "\n", "3. **con_agent**：反方智能体\n", " - 根据当前轮次给出反对观点\n", " - 将论点添加到辩论历史\n", "\n", "4. **judge_agent**：裁判智能体\n", " - 总结所有辩论内容\n", " - 给出最终结论\n", "\n", "5. **after_con_route**：反方发言后的路由\n", " - 如果还没到第2轮，回到正方开始下一轮\n", " - 如果已经第2轮结束，进入裁判\n", "\n", "6. **图结构**：\n", " - START -> pro -> con\n", " - con 根据轮次决定回到 pro 还是进入 judge\n", " - judge -> END\n", "\n", "流程图：\n", "\n", "```\n", "START -> pro -> con\n", " │\n", " ┌────────┴────────┐\n", " ▼ ▼\n", " pro → con judge -> END\n", "```\n", "\n", "这种模式的优点：\n", "- 可以全面分析问题的正反两面\n", "- 通过辩论深入探讨复杂问题\n", "- 最终结论更加客观全面" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 7. 综合案例：多智能体协作完成复杂任务\n", "\n", "现在我们将多种模式结合起来，构建一个复杂的多智能体系统。\n", "\n", "场景：\n", "- 用户提出一个数据分析需求\n", "- 监督者智能体负责协调\n", "- 数据收集智能体收集数据\n", "- 数据处理智能体处理数据\n", "- 数据分析智能体分析数据\n", "- 报告生成智能体生成报告\n", "- 使用MemorySaver保存状态" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing_extensions import TypedDict\n", "from langgraph.graph import StateGraph, START, END\n", "from langgraph.checkpoint.memory import MemorySaver\n", "from openai import OpenAI\n", "import os\n", "from dotenv import load_dotenv\n", "\n", "load_dotenv()\n", "\n", "client = OpenAI(\n", " base_url=os.getenv(\"OPENAI_BASE_URL\"),\n", " api_key=os.getenv(\"OPENAI_API_KEY\")\n", ")\n", "\n", "def call_llm(system_prompt: str, user_prompt: str) -> str:\n", " response = client.chat.completions.create(\n", " model=\"qwen3.6-35b-A3b\",\n", " messages=[\n", " {\"role\": \"system\", \"content\": system_prompt},\n", " {\"role\": \"user\", \"content\": user_prompt}\n", " ],\n", " temperature=0.7,\n", " max_tokens=500\n", " )\n", " return response.choices[0].message.content\n", "\n", "class DataAgentState(TypedDict):\n", " user_request: str\n", " raw_data: str\n", " processed_data: str\n", " analysis_result: str\n", " report: str\n", " next_agent: str\n", "\n", "def supervisor(state: DataAgentState):\n", " print(f'监督者：收到请求：{state[\"user_request\"]}')\n", " \n", " if not state['raw_data']:\n", " print('监督者：分配给数据收集智能体')\n", " return {'next_agent': 'collect'}\n", " elif not state['processed_data']:\n", " print('监督者：分配给数据处理智能体')\n", " return {'next_agent': 'process'}\n", " elif not state['analysis_result']:\n", " print('监督者：分配给数据分析智能体')\n", " return {'next_agent': 'analyze'}\n", " elif not state['report']:\n", " print('监督者：分配给报告生成智能体')\n", " return {'next_agent': 'report'}\n", " else:\n", " print('监督者：任务完成')\n", " return {'next_agent': 'finish'}\n", "\n", "def collect_agent(state: DataAgentState):\n", " print(f'数据收集智能体：正在收集数据')\n", " raw_data = f'原始数据：关于\"{state[\"user_request\"]}\"的统计数据、趋势数据、对比数据。'\n", " print(f'数据收集智能体：完成数据收集')\n", " return {'raw_data': raw_data}\n", "\n", "def process_agent(state: DataAgentState):\n", " print(f'数据处理智能体：正在处理数据')\n", " processed_data = f'处理后数据：清洗、转换、标准化后的\"{state[\"user_request\"]}\"数据。'\n", " print(f'数据处理智能体：完成数据处理')\n", " return {'processed_data': processed_data}\n", "\n", "def analyze_agent(state: DataAgentState):\n", " print(f'数据分析智能体：正在分析数据')\n", " \n", " system_prompt = '你是一个专业的数据分析师，请根据提供的数据进行深入分析。'\n", " user_prompt = f'请分析以下数据，找出关键发现和趋势：\\n\\n原始数据：{state[\"raw_data\"]}\\n\\n处理后数据：{state[\"processed_data\"]}'\n", " \n", " analysis_result = call_llm(system_prompt, user_prompt)\n", " \n", " print(f'数据分析智能体：完成数据分析')\n", " return {'analysis_result': analysis_result}\n", "\n", "def report_agent(state: DataAgentState):\n", " print(f'报告生成智能体：正在生成报告')\n", " \n", " system_prompt = '你是一个专业的报告撰写者，请根据数据分析结果撰写结构化的报告。'\n", " user_prompt = f'请根据以下信息撰写一份完整的数据分析报告：\\n\\n请求：{state[\"user_request\"]}\\n\\n分析结果：{state[\"analysis_result\"]}'\n", " \n", " report = call_llm(system_prompt, user_prompt)\n", " print(f'报告生成智能体：完成报告生成')\n", " return {'report': report}\n", "\n", "def route(state: DataAgentState):\n", " if state['next_agent'] == 'collect':\n", " return 'collect'\n", " elif state['next_agent'] == 'process':\n", " return 'process'\n", " elif state['next_agent'] == 'analyze':\n", " return 'analyze'\n", " elif state['next_agent'] == 'report':\n", " return 'report'\n", " elif state['next_agent'] == 'finish':\n", " return END\n", " return 'supervisor'\n", "\n", "builder = StateGraph(DataAgentState)\n", "builder.add_node('supervisor', supervisor)\n", "builder.add_node('collect', collect_agent)\n", "builder.add_node('process', process_agent)\n", "builder.add_node('analyze', analyze_agent)\n", "builder.add_node('report', report_agent)\n", "\n", "builder.add_edge(START, 'supervisor')\n", "builder.add_edge('collect', 'supervisor')\n", "builder.add_edge('process', 'supervisor')\n", "builder.add_edge('analyze', 'supervisor')\n", "builder.add_edge('report', 'supervisor')\n", "builder.add_conditional_edges(\n", " 'supervisor',\n", " route,\n", " {\n", " 'collect': 'collect',\n", " 'process': 'process',\n", " 'analyze': 'analyze',\n", " 'report': 'report',\n", " END: END\n", " }\n", ")\n", "\n", "memory = MemorySaver()\n", "graph = builder.compile(checkpointer=memory)\n", "\n", "config = {'configurable': {'thread_id': 'data_analysis_session'}}\n", "\n", "result = graph.invoke({\n", " 'user_request': '分析过去一年的销售趋势',\n", " 'raw_data': '',\n", " 'processed_data': '',\n", " 'analysis_result': '',\n", " 'report': '',\n", " 'next_agent': ''\n", "}, config)\n", "\n", "print()\n", "print('最终报告：')\n", "print(result['report'])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这个案例展示了一个完整的多智能体协作系统：\n", "\n", "1. **环境配置**：\n", " - 加载 `.env` 文件中的 API 密钥\n", " - 创建 OpenAI 客户端\n", " - 定义 `call_llm` 函数，封装大模型调用逻辑\n", "\n", "2. **DataAgentState**：定义状态，包含六个字段，分别对应每个阶段的数据。\n", "\n", "3. **supervisor**：监督者智能体\n", " - 根据当前状态决定下一步分配给哪个智能体\n", " - 按照数据收集→数据处理→数据分析→报告生成的顺序分配\n", "\n", "4. **collect_agent**：数据收集智能体\n", " - 收集原始数据\n", " - 保存到 `raw_data`\n", "\n", "5. **process_agent**：数据处理智能体\n", " - 处理原始数据\n", " - 保存到 `processed_data`\n", "\n", "6. **analyze_agent**：数据分析智能体（使用真实大模型）\n", " - 构建系统提示词和用户提示词\n", " - 调用大模型分析数据\n", " - 保存到 `analysis_result`\n", "\n", "7. **report_agent**：报告生成智能体（使用真实大模型）\n", " - 构建系统提示词和用户提示词\n", " - 调用大模型生成报告\n", " - 保存到 `report`\n", "\n", "8. **MemorySaver**：保存状态\n", " - 使用 `thread_id` 隔离不同会话\n", " - 支持多轮交互\n", "\n", "流程图：\n", "\n", "```\n", "START -> supervisor\n", " │\n", " ┌──────┼──────┬──────┬──────┐\n", " ▼ ▼ ▼ ▼ ▼\n", " collect process analyze report finish\n", " │ │ │ │\n", " └──────┴──────┴──────┘\n", " │\n", " ▼\n", " supervisor -> END\n", "```\n", "\n", "这个系统结合了监督者模式和流水线模式，展示了多智能体协作的强大能力。" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 8. 总结\n", "\n", "### 核心知识点\n", "\n", "1. **监督者模式**：一个监督者智能体协调多个工作智能体\n", "2. **协作模式**：多个智能体平等协作，共同完成任务\n", "3. **竞争模式**：多个智能体竞争，选择最优解\n", "4. **流水线模式**：多个智能体按顺序处理任务\n", "5. **辩论模式**：多个智能体展开辩论，最终达成共识\n", "\n", "### 架构选择指南\n", "\n", "| 模式 | 适用场景 | 特点 |\n", "| --- | --- | --- |\n", "| 监督者 | 需要多种专业能力 | 集中控制，易于管理 |\n", "| 协作 | 需要频繁交流 | 平等协作，灵活适应 |\n", "| 竞争 | 需要最优解 | 多方竞争，提高质量 |\n", "| 流水线 | 可分解为顺序步骤 | 结构清晰，易于扩展 |\n", "| 辩论 | 需要深入分析 | 全面探讨，结论客观 |\n", "\n", "### 实践要点\n", "\n", "- 根据任务特点选择合适的架构模式\n", "- 智能体之间通过共享状态传递信息\n", "- 使用监督者模式时，确保监督者逻辑清晰\n", "- 使用MemorySaver保存状态，支持多轮交互\n", "- 可以组合多种模式构建复杂系统\n", "- 智能体可以调用真实大模型（如GPT、Qwen等）来生成内容\n", "- 使用环境变量管理API密钥，确保安全性\n", "- 通过封装call_llm函数统一管理大模型调用" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 练习\n", "\n", "1. 修改监督者模式示例，添加一个\"质量检查智能体\"\n", "2. 修改流水线模式示例，添加一个\"校对智能体\"\n", "3. 修改竞争模式示例，让智能体可以互相评价\n", "4. 创建一个新的辩论模式示例，增加到3轮辩论\n", "5. 设计一个综合多智能体系统，包含监督者、协作和竞争模式" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.14.4" } }, "nbformat": 4, "nbformat_minor": 4 }