{ "cells": [ { "cell_type": "markdown", "id": "2046debc", "metadata": {}, "source": [ "# 阶段3：任务分解与多步推理 Agent 实践\n", "\n", "本笔记本帮助你实践任务分解、ReAct 模式和多步推理等内容，参考 docs/stage3-multi-step.md。\n", "\n", "## 本阶段目标\n", "- 理解任务分解的原理和方法\n", "- 掌握 ReAct 模式的实现\n", "- 学习让 Agent 进行多步骤推理\n", "- 实践构建能够自主规划的工作流 Agent" ] }, { "cell_type": "markdown", "id": "e82e57f2", "metadata": {}, "source": [ "## 1. 环境配置" ] }, { "cell_type": "code", "execution_count": null, "id": "b92b7d15", "metadata": {}, "outputs": [], "source": [ "# 导入必要的库\n", "import os\n", "import json\n", "import re\n", "from datetime import datetime\n", "from typing import List, Dict, Any\n", "from dotenv import load_dotenv\n", "from openai import AzureOpenAI\n", "\n", "# 加载环境变量\n", "load_dotenv(override=True)\n", "\n", "# 创建 Azure OpenAI 客户端\n", "client = AzureOpenAI(\n", " azure_endpoint=os.getenv(\"AZURE_OPENAI_ENDPOINT\"),\n", " api_key=os.getenv(\"AZURE_OPENAI_API_KEY\"),\n", " api_version=os.getenv(\"AZURE_OPENAI_API_VERSION\", \"2024-12-01-preview\")\n", ")\n", "\n", "# 获取部署名称\n", "deployment = os.getenv(\"AZURE_OPENAI_DEPLOYMENT\", \"gpt-4o\")\n", "\n", "print(f\"✅ 环境配置完成\")\n", "print(f\"📌 使用部署: {deployment}\")" ] }, { "cell_type": "markdown", "id": "d9d1a349", "metadata": {}, "source": [ "## 2. 任务分解核心概念\n", "\n", "复杂任务通常无法一步完成，需要：\n", "1. **分解**：将大任务拆解为小步骤\n", "2. **规划**：确定执行顺序\n", "3. **执行**：逐步完成各个子任务\n", "4. **整合**：汇总结果\n", "\n", "### 任务分解模式\n", "\n", "```\n", "线性分解: 任务 → 步骤1 → 步骤2 → 步骤3 → 结果\n", "\n", "并行分解: 任务 → ├─ 子任务A ─┤\n", " ├─ 子任务B ─┼→ 整合 → 结果\n", " └─ 子任务C ─┘\n", "\n", "层级分解: 任务 → 子任务1 → 子子任务1.1\n", " → 子子任务1.2\n", " → 子任务2 → 子子任务2.1\n", "```" ] }, { "cell_type": "markdown", "id": "447865c0", "metadata": {}, "source": [ "## 3. ReAct 模式实现\n", "\n", "ReAct (Reasoning + Acting) 是一种让 LLM 交替进行推理和行动的模式：\n", "\n", "```\n", "Thought: [思考当前情况]\n", "Action: [决定采取的行动]\n", "Observation: [观察行动结果]\n", "... (循环)\n", "Final Answer: [最终答案]\n", "```" ] }, { "cell_type": "code", "execution_count": null, "id": "476797b4", "metadata": {}, "outputs": [], "source": [ "class ReActAgent:\n", " \"\"\"ReAct 模式 Agent\n", " \n", " 实现 Reasoning + Acting 循环，让 LLM 能够：\n", " 1. 思考当前情况\n", " 2. 选择合适的工具执行\n", " 3. 观察结果\n", " 4. 继续推理直到得出答案\n", " \"\"\"\n", " \n", " def __init__(self, max_iterations=10):\n", " self.max_iterations = max_iterations\n", " self.tools = self._define_tools()\n", " \n", " def _define_tools(self):\n", " \"\"\"定义可用工具\"\"\"\n", " return {\n", " \"search\": self.search_tool,\n", " \"calculate\": self.calculate_tool,\n", " \"lookup\": self.lookup_tool,\n", " }\n", " \n", " def search_tool(self, query: str) -> str:\n", " \"\"\"搜索工具（模拟）\n", " \n", " 实际应用中应调用真实搜索API，如：\n", " - Bing Search API\n", " - Google Custom Search\n", " - Wikipedia API\n", " \"\"\"\n", " mock_results = {\n", " \"Python\": \"Python是一种高级编程语言，由Guido van Rossum创建于1991年。它以简洁和可读性著称。\",\n", " \"北京\": \"北京是中国的首都，人口超过2000万，是中国的政治、文化中心。\",\n", " \"AI\": \"人工智能(AI)是计算机科学的分支，研究如何让机器模拟人类智能。当前热门领域包括深度学习和大语言模型。\",\n", " \"Agent\": \"AI Agent是能够自主感知环境、做出决策并采取行动的智能系统。\",\n", " \"ReAct\": \"ReAct是一种结合推理(Reasoning)和行动(Acting)的提示工程技术，让LLM能够交替思考和执行工具。\"\n", " }\n", " for key in mock_results:\n", " if key.lower() in query.lower():\n", " return mock_results[key]\n", " return f\"关于'{query}'的搜索结果：暂无相关信息。\"\n", " \n", " def calculate_tool(self, expression: str) -> str:\n", " \"\"\"计算工具\n", " \n", " 安全地执行数学表达式计算\n", " \"\"\"\n", " try:\n", " # 限制只允许基本数学运算\n", " allowed_chars = set(\"0123456789+-*/().% \")\n", " if not all(c in allowed_chars for c in expression):\n", " return \"计算错误：表达式包含不允许的字符\"\n", " result = eval(expression, {\"__builtins__\": {}}, {})\n", " return str(result)\n", " except Exception as e:\n", " return f\"计算错误: {e}\"\n", " \n", " def lookup_tool(self, keyword: str) -> str:\n", " \"\"\"查找工具（在知识库中查找）\"\"\"\n", " knowledge_base = {\n", " \"GPT\": \"GPT (Generative Pre-trained Transformer) 是OpenAI开发的大语言模型系列。\",\n", " \"LLM\": \"LLM (Large Language Model) 大语言模型，通过大规模文本训练获得语言理解和生成能力。\"\n", " }\n", " for key in knowledge_base:\n", " if key.lower() in keyword.lower():\n", " return knowledge_base[key]\n", " return f\"在知识库中未找到关于'{keyword}'的信息\"\n", " \n", " def parse_action(self, text: str) -> tuple:\n", " \"\"\"解析 LLM 输出的行动\n", " \n", " 匹配格式: Action: tool_name[arguments]\n", " \"\"\"\n", " action_pattern = r\"Action:\\s*(\\w+)\\[(.*)\\]\"\n", " match = re.search(action_pattern, text)\n", " \n", " if match:\n", " tool_name = match.group(1)\n", " arguments = match.group(2).strip('\"\\'')\n", " return tool_name, arguments\n", " return None, None\n", " \n", " def run(self, question: str) -> Dict:\n", " \"\"\"运行 ReAct 循环\n", " \n", " Args:\n", " question: 用户问题\n", " \n", " Returns:\n", " 包含答案、推理过程和迭代次数的字典\n", " \"\"\"\n", " prompt = f\"\"\"你是一个使用 ReAct 模式的智能助手。\n", "\n", "可用工具：\n", "- search[query]: 搜索信息，参数是搜索关键词\n", "- calculate[expression]: 计算数学表达式\n", "- lookup[keyword]: 在知识库中查找关键词\n", "\n", "请按以下格式思考和行动：\n", "Thought: [你的思考过程]\n", "Action: [工具名称][参数]\n", "Observation: [工具返回的结果，由系统填充]\n", "... (可以重复 Thought/Action/Observation)\n", "Thought: 我现在知道最终答案了\n", "Final Answer: [最终答案]\n", "\n", "重要规则：\n", "1. 每次只执行一个 Action\n", "2. 等待 Observation 后再继续\n", "3. 当有足够信息时给出 Final Answer\n", "\n", "问题: {question}\n", "\n", "开始！\n", "Thought:\"\"\"\n", " \n", " messages = [{\"role\": \"user\", \"content\": prompt}]\n", " full_response = \"\"\n", " \n", " for i in range(self.max_iterations):\n", " response = client.chat.completions.create(\n", " model=deployment,\n", " messages=messages,\n", " max_tokens=500\n", " )\n", " \n", " thought = response.choices[0].message.content\n", " full_response += thought + \"\\n\"\n", " \n", " # 检查是否得到最终答案\n", " if \"Final Answer:\" in thought:\n", " final_answer = thought.split(\"Final Answer:\")[-1].strip()\n", " return {\n", " \"answer\": final_answer,\n", " \"process\": full_response,\n", " \"iterations\": i + 1\n", " }\n", " \n", " # 解析行动\n", " tool_name, arguments = self.parse_action(thought)\n", " \n", " if tool_name and tool_name in self.tools:\n", " # 执行工具\n", " observation = self.tools[tool_name](arguments)\n", " full_response += f\"Observation: {observation}\\n\"\n", " \n", " # 更新消息\n", " messages.append({\"role\": \"assistant\", \"content\": thought})\n", " messages.append({\n", " \"role\": \"user\",\n", " \"content\": f\"Observation: {observation}\\nThought:\"\n", " })\n", " else:\n", " # 如果没有找到有效的行动，提示继续\n", " messages.append({\"role\": \"assistant\", \"content\": thought})\n", " messages.append({\n", " \"role\": \"user\",\n", " \"content\": \"请继续思考或给出最终答案。\\nThought:\"\n", " })\n", " \n", " return {\n", " \"answer\": \"达到最大迭代次数，未能得到答案\",\n", " \"process\": full_response,\n", " \"iterations\": self.max_iterations\n", " }\n", "\n", "print(\"✅ ReActAgent 类定义完成\")" ] }, { "cell_type": "markdown", "id": "54145e1b", "metadata": {}, "source": [ "### 3.1 测试 ReAct Agent" ] }, { "cell_type": "code", "execution_count": null, "id": "a6dd783b", "metadata": {}, "outputs": [], "source": [ "# 创建 ReAct Agent 实例\n", "react_agent = ReActAgent(max_iterations=5)\n", "\n", "# 测试问题：需要搜索才能回答的问题\n", "question = \"Python语言是什么时候创建的？创建者是谁？\"\n", "\n", "print(\"=\" * 60)\n", "print(f\"❓ 问题: {question}\")\n", "print(\"=\" * 60)\n", "\n", "result = react_agent.run(question)\n", "\n", "print(\"\\n📝 推理过程:\")\n", "print(\"-\" * 60)\n", "print(result[\"process\"])\n", "print(\"=\" * 60)\n", "print(f\"✅ 最终答案: {result['answer']}\")\n", "print(f\"🔄 迭代次数: {result['iterations']}\")" ] }, { "cell_type": "code", "execution_count": null, "id": "88e9c758", "metadata": {}, "outputs": [], "source": [ "# 测试另一个问题：需要计算的问题\n", "question = \"如果我买了3个苹果，每个5元，又买了2个橙子，每个3元，一共花了多少钱？\"\n", "\n", "print(\"=\" * 60)\n", "print(f\"❓ 问题: {question}\")\n", "print(\"=\" * 60)\n", "\n", "result = react_agent.run(question)\n", "\n", "print(\"\\n📝 推理过程:\")\n", "print(\"-\" * 60)\n", "print(result[\"process\"])\n", "print(\"=\" * 60)\n", "print(f\"✅ 最终答案: {result['answer']}\")\n", "print(f\"🔄 迭代次数: {result['iterations']}\")" ] }, { "cell_type": "markdown", "id": "5e9f641e", "metadata": {}, "source": [ "## 4. 自动研究 Agent\n", "\n", "这个 Agent 能够：\n", "1. 将研究主题分解为子问题\n", "2. 逐个研究每个子问题\n", "3. 综合所有结果生成报告" ] }, { "cell_type": "code", "execution_count": null, "id": "e169219a", "metadata": {}, "outputs": [], "source": [ "class ResearchAgent:\n", " \"\"\"自动研究和资料收集 Agent\n", " \n", " 实现自动化研究流程：\n", " 1. 任务分解 - 将研究主题分解为子问题\n", " 2. 逐项研究 - 针对每个子问题收集信息\n", " 3. 综合报告 - 整合所有发现生成完整报告\n", " \"\"\"\n", " \n", " def __init__(self):\n", " self.research_results = []\n", " \n", " def decompose_research_task(self, topic: str) -> List[str]:\n", " \"\"\"分解研究任务为子问题\n", " \n", " Args:\n", " topic: 研究主题\n", " \n", " Returns:\n", " 子问题列表\n", " \"\"\"\n", " prompt = f\"\"\"研究主题: {topic}\n", "\n", "请将这个研究主题分解为3-5个具体的子问题，这些问题的答案能够全面覆盖该主题。\n", "\n", "要求：\n", "1. 每个子问题应该是具体、可回答的\n", "2. 子问题之间应该互补，避免重复\n", "3. 涵盖主题的不同方面\n", "\n", "以JSON格式返回：\n", "{{\n", " \"subtopics\": [\n", " \"子问题1\",\n", " \"子问题2\",\n", " \"子问题3\"\n", " ]\n", "}}\"\"\"\n", " \n", " response = client.chat.completions.create(\n", " model=deployment,\n", " messages=[{\"role\": \"user\", \"content\": prompt}],\n", " response_format={\"type\": \"json_object\"}\n", " )\n", " \n", " result = json.loads(response.choices[0].message.content)\n", " return result.get(\"subtopics\", [])\n", " \n", " def research_subtopic(self, subtopic: str) -> Dict:\n", " \"\"\"研究单个子主题\n", " \n", " 在实际应用中，这里应该：\n", " - 调用搜索API获取相关信息\n", " - 访问知识库或数据库\n", " - 调用专业API（如学术论文API）\n", " \"\"\"\n", " prompt = f\"\"\"请简要回答以下问题（100-200字）：\n", "{subtopic}\n", "\n", "请给出准确、专业的回答，包含关键信息和要点。\"\"\"\n", " \n", " response = client.chat.completions.create(\n", " model=deployment,\n", " messages=[{\"role\": \"user\", \"content\": prompt}]\n", " )\n", " \n", " answer = response.choices[0].message.content\n", " \n", " return {\n", " \"question\": subtopic,\n", " \"answer\": answer,\n", " \"source\": \"Azure OpenAI\"\n", " }\n", " \n", " def synthesize_results(self, topic: str, results: List[Dict]) -> str:\n", " \"\"\"综合研究结果生成报告\"\"\"\n", " results_text = \"\\n\\n\".join([\n", " f\"### {i+1}. {r['question']}\\n{r['answer']}\"\n", " for i, r in enumerate(results)\n", " ])\n", " \n", " prompt = f\"\"\"研究主题: {topic}\n", "\n", "以下是对该主题的分项研究结果：\n", "\n", "{results_text}\n", "\n", "请将这些信息综合成一份连贯、结构化的研究报告（300-500字），包括：\n", "1. **概述** - 主题简介\n", "2. **主要发现** - 核心内容\n", "3. **总结** - 关键结论\n", "\n", "使用 Markdown 格式。\"\"\"\n", " \n", " response = client.chat.completions.create(\n", " model=deployment,\n", " messages=[{\"role\": \"user\", \"content\": prompt}]\n", " )\n", " \n", " return response.choices[0].message.content\n", " \n", " def conduct_research(self, topic: str) -> Dict:\n", " \"\"\"执行完整的研究流程\"\"\"\n", " print(f\"📚 开始研究: {topic}\\n\")\n", " print(\"=\" * 60)\n", " \n", " # 步骤1: 任务分解\n", " print(\"📋 步骤1: 分解研究任务...\")\n", " subtopics = self.decompose_research_task(topic)\n", " print(f\"已分解为 {len(subtopics)} 个子问题：\")\n", " for i, st in enumerate(subtopics, 1):\n", " print(f\" {i}. {st}\")\n", " print()\n", " \n", " # 步骤2: 逐个研究子主题\n", " print(\"🔍 步骤2: 研究各个子主题...\")\n", " results = []\n", " for i, subtopic in enumerate(subtopics, 1):\n", " print(f\" 研究中 ({i}/{len(subtopics)}): {subtopic[:40]}...\")\n", " result = self.research_subtopic(subtopic)\n", " results.append(result)\n", " print()\n", " \n", " # 步骤3: 综合报告\n", " print(\"📝 步骤3: 生成综合报告...\")\n", " report = self.synthesize_results(topic, results)\n", " print(\"完成！\\n\")\n", " print(\"=\" * 60)\n", " \n", " return {\n", " \"topic\": topic,\n", " \"subtopics\": subtopics,\n", " \"detailed_results\": results,\n", " \"final_report\": report\n", " }\n", "\n", "print(\"✅ ResearchAgent 类定义完成\")" ] }, { "cell_type": "markdown", "id": "1c77c4b7", "metadata": {}, "source": [ "### 4.1 测试研究 Agent" ] }, { "cell_type": "code", "execution_count": null, "id": "cd98b2af", "metadata": {}, "outputs": [], "source": [ "# 创建研究 Agent\n", "research_agent = ResearchAgent()\n", "\n", "# 执行研究\n", "topic = \"AI Agent的核心技术\"\n", "result = research_agent.conduct_research(topic)" ] }, { "cell_type": "code", "execution_count": null, "id": "ef6079ae", "metadata": {}, "outputs": [], "source": [ "# 显示研究报告\n", "print(\"📊 研究报告\")\n", "print(\"=\" * 60)\n", "print(result[\"final_report\"])" ] }, { "cell_type": "code", "execution_count": null, "id": "e15307c4", "metadata": {}, "outputs": [], "source": [ "# 查看详细研究结果\n", "print(\"📋 详细研究结果\")\n", "print(\"=\" * 60)\n", "for i, r in enumerate(result[\"detailed_results\"], 1):\n", " print(f\"\\n【问题 {i}】{r['question']}\")\n", " print(f\"【回答】{r['answer']}\")\n", " print(\"-\" * 40)" ] }, { "cell_type": "markdown", "id": "5833efaa", "metadata": {}, "source": [ "## 5. 工作流自动化 Agent\n", "\n", "这个 Agent 能够：\n", "1. 理解用户的工作请求\n", "2. 自动规划执行步骤\n", "3. 按顺序执行各个操作\n", "4. 汇报执行结果" ] }, { "cell_type": "code", "execution_count": null, "id": "568575dd", "metadata": {}, "outputs": [], "source": [ "class WorkflowAgent:\n", " \"\"\"工作流自动化助手\n", " \n", " 能够理解用户请求，自动规划和执行工作流程。\n", " \"\"\"\n", " \n", " def __init__(self):\n", " self.workflow_history = []\n", " self.available_actions = {\n", " \"send_email\": self.send_email,\n", " \"create_task\": self.create_task,\n", " \"schedule_meeting\": self.schedule_meeting,\n", " \"generate_document\": self.generate_document,\n", " \"analyze_data\": self.analyze_data\n", " }\n", " \n", " def send_email(self, recipient: str, subject: str, body: str = \"\") -> Dict:\n", " \"\"\"发送邮件（模拟）\"\"\"\n", " print(f\" 📧 发送邮件到: {recipient}\")\n", " print(f\" 主题: {subject}\")\n", " return {\"status\": \"sent\", \"timestamp\": datetime.now().isoformat()}\n", " \n", " def create_task(self, title: str, description: str = \"\", assignee: str = \"\") -> Dict:\n", " \"\"\"创建任务（模拟）\"\"\"\n", " task_id = f\"TASK-{len(self.workflow_history)+1:03d}\"\n", " print(f\" ✅ 创建任务: {title}\")\n", " print(f\" ID: {task_id}, 负责人: {assignee or '未指定'}\")\n", " return {\"task_id\": task_id, \"status\": \"created\"}\n", " \n", " def schedule_meeting(self, title: str, participants: str = \"\", duration: int = 60) -> Dict:\n", " \"\"\"安排会议（模拟）\"\"\"\n", " meeting_id = f\"MEET-{len(self.workflow_history)+1:03d}\"\n", " print(f\" 📅 安排会议: {title}\")\n", " print(f\" ID: {meeting_id}, 时长: {duration}分钟\")\n", " return {\"meeting_id\": meeting_id, \"status\": \"scheduled\"}\n", " \n", " def generate_document(self, doc_type: str, content: str = \"\") -> Dict:\n", " \"\"\"生成文档（模拟）\"\"\"\n", " doc_id = f\"DOC-{len(self.workflow_history)+1:03d}\"\n", " print(f\" 📄 生成文档: {doc_type}\")\n", " print(f\" ID: {doc_id}\")\n", " return {\"doc_id\": doc_id, \"status\": \"generated\"}\n", " \n", " def analyze_data(self, data_source: str, analysis_type: str = \"general\") -> Dict:\n", " \"\"\"分析数据（模拟）\"\"\"\n", " analysis_id = f\"ANAL-{len(self.workflow_history)+1:03d}\"\n", " print(f\" 📊 分析数据: {data_source}\")\n", " print(f\" ID: {analysis_id}, 类型: {analysis_type}\")\n", " return {\"analysis_id\": analysis_id, \"status\": \"completed\"}\n", " \n", " def plan_workflow(self, user_request: str) -> List[Dict]:\n", " \"\"\"规划工作流步骤\"\"\"\n", " prompt = f\"\"\"用户请求: {user_request}\n", "\n", "可用操作:\n", "1. send_email - 发送邮件，参数: recipient(收件人), subject(主题), body(正文)\n", "2. create_task - 创建任务，参数: title(标题), description(描述), assignee(负责人)\n", "3. schedule_meeting - 安排会议，参数: title(标题), participants(参与者), duration(时长分钟)\n", "4. generate_document - 生成文档，参数: doc_type(文档类型), content(内容描述)\n", "5. analyze_data - 分析数据，参数: data_source(数据源), analysis_type(分析类型)\n", "\n", "请将用户请求分解为具体的操作步骤，以JSON格式返回：\n", "{{\n", " \"steps\": [\n", " {{\n", " \"step_number\": 1,\n", " \"action\": \"操作名称\",\n", " \"parameters\": {{\"param1\": \"value1\"}},\n", " \"description\": \"步骤描述\"\n", " }}\n", " ]\n", "}}\n", "\n", "注意：\n", "- 步骤要有逻辑顺序\n", "- 参数要具体明确\n", "- 每个步骤要有清晰的描述\"\"\"\n", " \n", " response = client.chat.completions.create(\n", " model=deployment,\n", " messages=[{\"role\": \"user\", \"content\": prompt}],\n", " response_format={\"type\": \"json_object\"}\n", " )\n", " \n", " plan = json.loads(response.choices[0].message.content)\n", " return plan.get(\"steps\", [])\n", " \n", " def execute_workflow(self, user_request: str) -> Dict:\n", " \"\"\"执行完整工作流\"\"\"\n", " print(f\"🚀 开始执行工作流\")\n", " print(f\"📝 用户请求: {user_request}\\n\")\n", " print(\"=\" * 60)\n", " \n", " # 步骤1: 规划\n", " print(\"📋 步骤1: 规划工作流\")\n", " print(\"-\" * 60)\n", " steps = self.plan_workflow(user_request)\n", " \n", " print(f\"已规划 {len(steps)} 个步骤：\")\n", " for step in steps:\n", " print(f\" {step['step_number']}. {step['description']}\")\n", " print()\n", " \n", " # 步骤2: 执行\n", " print(\"=\" * 60)\n", " print(\"⚙️ 步骤2: 执行工作流\")\n", " print(\"-\" * 60)\n", " \n", " execution_results = []\n", " for step in steps:\n", " print(f\"\\n执行步骤 {step['step_number']}: {step['description']}\")\n", " \n", " action_name = step['action']\n", " if action_name in self.available_actions:\n", " try:\n", " action_func = self.available_actions[action_name]\n", " result = action_func(**step['parameters'])\n", " \n", " execution_results.append({\n", " \"step\": step['step_number'],\n", " \"action\": action_name,\n", " \"status\": \"success\",\n", " \"result\": result\n", " })\n", " self.workflow_history.append(result)\n", " print(f\" ✓ 完成\")\n", " except Exception as e:\n", " execution_results.append({\n", " \"step\": step['step_number'],\n", " \"action\": action_name,\n", " \"status\": \"failed\",\n", " \"error\": str(e)\n", " })\n", " print(f\" ✗ 失败: {e}\")\n", " else:\n", " print(f\" ⚠ 未知操作: {action_name}\")\n", " execution_results.append({\n", " \"step\": step['step_number'],\n", " \"action\": action_name,\n", " \"status\": \"skipped\",\n", " \"error\": \"未知操作\"\n", " })\n", " \n", " print(\"\\n\" + \"=\" * 60)\n", " print(\"✅ 工作流执行完成\")\n", " print(\"=\" * 60)\n", " \n", " # 统计\n", " success_count = sum(1 for r in execution_results if r['status'] == 'success')\n", " failed_count = sum(1 for r in execution_results if r['status'] == 'failed')\n", " \n", " return {\n", " \"request\": user_request,\n", " \"planned_steps\": steps,\n", " \"execution_results\": execution_results,\n", " \"summary\": {\n", " \"total\": len(steps),\n", " \"success\": success_count,\n", " \"failed\": failed_count\n", " }\n", " }\n", "\n", "print(\"✅ WorkflowAgent 类定义完成\")" ] }, { "cell_type": "markdown", "id": "15b91362", "metadata": {}, "source": [ "### 5.1 测试工作流 Agent" ] }, { "cell_type": "code", "execution_count": null, "id": "bf94c773", "metadata": {}, "outputs": [], "source": [ "# 创建工作流 Agent\n", "workflow_agent = WorkflowAgent()\n", "\n", "# 用户请求\n", "request = \"\"\"\n", "我们需要启动一个新项目：\n", "1. 给团队成员发邮件通知项目启动\n", "2. 创建项目主要任务\n", "3. 安排项目启动会议\n", "4. 生成项目计划文档\n", "\"\"\"\n", "\n", "# 执行工作流\n", "result = workflow_agent.execute_workflow(request)" ] }, { "cell_type": "code", "execution_count": null, "id": "70eab61f", "metadata": {}, "outputs": [], "source": [ "# 查看执行摘要\n", "print(\"\\n📊 执行摘要:\")\n", "print(f\" 总步骤数: {result['summary']['total']}\")\n", "print(f\" 成功: {result['summary']['success']}\")\n", "print(f\" 失败: {result['summary']['failed']}\")" ] }, { "cell_type": "markdown", "id": "4ecb7815", "metadata": {}, "source": [ "### 5.2 测试另一个工作流" ] }, { "cell_type": "code", "execution_count": null, "id": "348cdd1c", "metadata": {}, "outputs": [], "source": [ "# 另一个工作流请求\n", "request2 = \"\"\"\n", "帮我完成月度报告的准备工作：\n", "- 分析本月销售数据\n", "- 生成月度报告文档\n", "- 发邮件给领导汇报\n", "\"\"\"\n", "\n", "result2 = workflow_agent.execute_workflow(request2)" ] }, { "cell_type": "markdown", "id": "5205d39f", "metadata": {}, "source": [ "## 6. 总结\n", "\n", "本阶段我们学习了：\n", "\n", "### 6.1 核心概念\n", "- **任务分解**：将复杂任务拆解为可执行的小步骤\n", "- **ReAct 模式**：交替进行推理(Thought)和行动(Action)\n", "- **多步推理**：通过多个步骤逐步解决问题\n", "\n", "### 6.2 实践项目\n", "1. **ReAct Agent**：能够思考、使用工具、观察结果的循环推理\n", "2. **Research Agent**：自动分解研究任务、收集信息、生成报告\n", "3. **Workflow Agent**：理解请求、规划步骤、自动执行\n", "\n", "### 6.3 关键技巧\n", "- 使用 JSON 格式确保 LLM 输出结构化内容\n", "- 通过正则表达式解析 LLM 输出中的动作\n", "- 维护对话历史实现多步推理\n", "- 错误处理和容错机制\n", "\n", "### 6.4 下一步\n", "进入 [阶段 4：记忆、知识库系统与长期状态](../docs/stage4-memory-rag.md)，学习如何赋予 Agent 记忆能力。" ] } ], "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.11.9" } }, "nbformat": 4, "nbformat_minor": 5 }