{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 第11章：周末出行清单智能体\n", "\n", "> 从单次工具调用到\"计划—行动—观察\"的多步循环\n", "\n", "本 Notebook 将带你构建一个最小智能体，让它自主完成多步任务：读取行程、计算天数、生成行李清单。\n", "\n", "**你将观察到**：\n", "- ReAct 循环怎样把单步变成多步\n", "- 状态机怎样跟踪进度\n", "- 步数上限怎样阻止无限循环\n", "- 人工暂停（HITL）怎样拦截高风险操作\n", "- 注入防御怎样保护系统安全\n", "\n", "**环境要求**：Python 3.10+，无需联网或外部模型（内置 mock 模式）" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 0. 环境准备" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import json\n", "import copy\n", "from datetime import datetime, timedelta\n", "from pathlib import Path\n", "from html import escape\n", "\n", "# 数据路径\n", "DATA_PATH = Path('../data/travel_data.json')\n", "\n", "print('环境检查通过 ✓')" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 加载旅行数据\n", "with open(DATA_PATH, 'r', encoding='utf-8') as f:\n", " travel_data = json.load(f)\n", "\n", "print(f'行程：{travel_data[\"itinerary\"][\"destination\"]}')\n", "print(f'出发日期：{travel_data[\"itinerary\"][\"start_date\"]}')\n", "print(f'返回日期：{travel_data[\"itinerary\"][\"end_date\"]}')\n", "print(f'天气天数：{len(travel_data[\"weather\"])}')\n", "print(f'允许工具：{[t[\"name\"] for t in travel_data[\"tools\"][\"allowed\"]]}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 1. 构建最小工具集\n", "\n", "智能体通过调用工具来完成任务。我们先定义5个模拟工具。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# ===== 模拟工具集 =====\n", "\n", "# 模拟文件系统\n", "virtual_fs = {\n", " 'itinerary.txt': json.dumps(travel_data['itinerary'], ensure_ascii=False, indent=2),\n", " 'weather.txt': json.dumps(travel_data['weather'], ensure_ascii=False, indent=2),\n", " 'traffic.txt': json.dumps(travel_data['traffic'], ensure_ascii=False, indent=2),\n", "}\n", "\n", "def read_text_file(file_path: str) -> str:\n", " \"\"\"读取虚拟文件\"\"\"\n", " filename = Path(file_path).name\n", " if filename in virtual_fs:\n", " return virtual_fs[filename]\n", " return f'[错误] 文件不存在: {filename}'\n", "\n", "def calc_days(start_date: str, end_date: str) -> int:\n", " \"\"\"计算出行天数\"\"\"\n", " d1 = datetime.strptime(start_date, '%Y-%m-%d')\n", " d2 = datetime.strptime(end_date, '%Y-%m-%d')\n", " return (d2 - d1).days + 1 # 包含首尾两天\n", "\n", "def create_note(filename: str, content: str) -> str:\n", " \"\"\"创建清单文件\"\"\"\n", " virtual_fs[filename] = content\n", " return f'[成功] 文件已保存: {filename}（{len(content)} 字符）'\n", "\n", "def list_files(directory: str = '.') -> list:\n", " \"\"\"列出虚拟文件\"\"\"\n", " return list(virtual_fs.keys())\n", "\n", "def delete_file(file_path: str) -> str:\n", " \"\"\"删除文件（高风险，需 HITL）\"\"\"\n", " filename = Path(file_path).name\n", " if filename in virtual_fs:\n", " del virtual_fs[filename]\n", " return f'[成功] 文件已删除: {filename}'\n", " return f'[错误] 文件不存在: {filename}'\n", "\n", "# 工具注册表\n", "TOOL_REGISTRY = {\n", " 'read_text_file': read_text_file,\n", " 'calc_days': calc_days,\n", " 'create_note': create_note,\n", " 'list_files': list_files,\n", " 'delete_file': delete_file,\n", "}\n", "\n", "# 风险级别\n", "RISK_LEVELS = {t['name']: t['risk_level'] for t in travel_data['tools']['allowed']}\n", "\n", "print('工具集构建完成 ✓')\n", "print(f'工具列表：{list(TOOL_REGISTRY.keys())}')\n", "print(f'风险级别：{RISK_LEVELS}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 2. 构建状态机\n", "\n", "状态机记录智能体的执行进度，决定是否可以继续。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "class AgentState:\n", " \"\"\"智能体状态机\"\"\"\n", " \n", " def __init__(self, max_steps=10):\n", " self.max_steps = max_steps\n", " self.current_step = 0\n", " self.status = 'pending' # pending/running/done/blocked/cancelled/failed\n", " self.step_log = [] # 每步的详细记录\n", " self.evidence = [] # 依据摘要\n", " self.hitl_pending = False # 是否有人工暂停等待处理\n", " self.hitl_request = None # 暂停请求详情\n", " \n", " def can_continue(self) -> bool:\n", " \"\"\"检查是否还能继续执行\"\"\"\n", " if self.status in ('done', 'cancelled', 'failed'):\n", " return False\n", " if self.hitl_pending:\n", " return False\n", " if self.current_step >= self.max_steps:\n", " self.status = 'blocked'\n", " return False\n", " return True\n", " \n", " def record_step(self, thought: str, action: str, observation: str, state_after: str):\n", " \"\"\"记录一步的执行\"\"\"\n", " self.current_step += 1\n", " entry = {\n", " 'step': self.current_step,\n", " 'thought': thought,\n", " 'action': action,\n", " 'observation': observation[:200] + '...' if len(observation) > 200 else observation,\n", " 'state_after': state_after,\n", " 'timestamp': datetime.now().isoformat()\n", " }\n", " self.step_log.append(entry)\n", " self.evidence.append(f'Step {self.current_step}: {action} → {state_after}')\n", " \n", " def request_hitl(self, tool_name: str, reason: str):\n", " \"\"\"请求人工确认\"\"\"\n", " self.hitl_pending = True\n", " self.status = 'blocked'\n", " self.hitl_request = {\n", " 'tool': tool_name,\n", " 'reason': reason,\n", " 'step': self.current_step + 1\n", " }\n", " \n", " def resolve_hitl(self, approved: bool) -> str:\n", " \"\"\"处理人工确认结果\"\"\"\n", " if not self.hitl_pending:\n", " return '无待处理的 HITL 请求'\n", " if approved:\n", " self.hitl_pending = False\n", " self.status = 'running'\n", " self.hitl_request['result'] = 'approved'\n", " return '已批准，继续执行'\n", " else:\n", " self.hitl_pending = False\n", " self.status = 'cancelled'\n", " self.hitl_request['result'] = 'cancelled'\n", " return '已取消，智能体停止'\n", " \n", " def summary(self) -> str:\n", " \"\"\"生成状态摘要\"\"\"\n", " return (\n", " f'状态: {self.status} | '\n", " f'步数: {self.current_step}/{self.max_steps} | '\n", " f'HITL: {\"等待中\" if self.hitl_pending else \"无\"}'\n", " )\n", "\n", "\n", "# 测试状态机\n", "state = AgentState(max_steps=5)\n", "print(f'初始状态: {state.summary()}')\n", "print(f'可以继续? {state.can_continue()}')\n", "\n", "state.record_step('需要读取行程', 'read_text_file(itinerary.txt)', '行程数据...', 'running')\n", "print(f'执行1步后: {state.summary()}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 3. 三步基线：读行程 → 算天数 → 建清单\n", "\n", "先运行最简单的3步任务，观察 ReAct 循环。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def run_baseline_agent(max_steps=10, include_weather=False):\n", " \"\"\"\n", " 模拟智能体的 ReAct 循环。\n", " \n", " 这里用预定义的步骤序列模拟\"模型决策\"。\n", " 真实场景中，每一步的\"思考\"和\"行动选择\"由大语言模型完成。\n", " \"\"\"\n", " state = AgentState(max_steps=max_steps)\n", " state.status = 'running'\n", " \n", " # 定义任务步骤序列\n", " steps_plan = [\n", " {\n", " 'thought': '我需要先读取行程信息，了解目的地和日期',\n", " 'tool': 'read_text_file',\n", " 'args': {'file_path': 'itinerary.txt'},\n", " },\n", " {\n", " 'thought': '我需要计算出行天数，以便决定带多少衣物',\n", " 'tool': 'calc_days',\n", " 'args': {\n", " 'start_date': travel_data['itinerary']['start_date'],\n", " 'end_date': travel_data['itinerary']['end_date']\n", " },\n", " },\n", " {\n", " 'thought': '现在我有了天数，可以生成基本行李清单',\n", " 'tool': 'create_note',\n", " 'args': {'filename': 'packing_list.txt', 'content': ''}, # content 动态生成\n", " },\n", " ]\n", " \n", " # 如果包含天气，增加步骤\n", " if include_weather:\n", " steps_plan.insert(2, {\n", " 'thought': '我还需要查看天气预报，决定是否需要雨具和防晒',\n", " 'tool': 'read_text_file',\n", " 'args': {'file_path': 'weather.txt'},\n", " })\n", " # 修改 create_note 步骤，加入天气信息\n", " steps_plan[-1]['thought'] = '现在我有了天数和天气，可以生成完整的行李清单'\n", " \n", " # 执行 ReAct 循环\n", " trip_days = None\n", " weather_info = None\n", " \n", " for step_def in steps_plan:\n", " if not state.can_continue():\n", " break\n", " \n", " thought = step_def['thought']\n", " tool_name = step_def['tool']\n", " args = step_def['args']\n", " \n", " # 检查 HITL\n", " if RISK_LEVELS.get(tool_name) == 'high':\n", " state.request_hitl(tool_name, f'高风险操作: {tool_name}')\n", " state.record_step(thought, f'[HITL] {tool_name}', '等待人工确认', 'blocked')\n", " break\n", " \n", " # 执行工具\n", " tool_func = TOOL_REGISTRY[tool_name]\n", " \n", " # 动态生成清单内容\n", " if tool_name == 'create_note':\n", " items = []\n", " items.append('=== 周末出行行李清单 ===')\n", " items.append(f'目的地：{travel_data[\"itinerary\"][\"destination\"]}')\n", " items.append(f'天数：{trip_days}天')\n", " items.append('')\n", " items.append('【证件】身份证、学生证')\n", " items.append(f'【衣物】内衣×{trip_days}套、袜子×{trip_days}双')\n", " items.append('【洗漱】牙刷、牙膏、毛巾、洗面奶')\n", " items.append('【电子】手机、充电器、耳机')\n", " items.append('【其他】纸巾、水杯、零食')\n", " if weather_info:\n", " items.append('')\n", " items.append('--- 天气相关 ---')\n", " has_rain = any(w['needs_rain_gear'] for w in weather_info)\n", " has_sun = any(w['needs_sunscreen'] for w in weather_info)\n", " if has_rain:\n", " items.append('【雨具】雨伞、防水外套')\n", " if has_sun:\n", " items.append('【防晒】防晒霜、遮阳帽、太阳镜')\n", " args['content'] = '\\n'.join(items)\n", " \n", " observation = str(tool_func(**args))\n", " \n", " # 保存中间结果\n", " if tool_name == 'calc_days':\n", " trip_days = int(observation)\n", " if tool_name == 'read_text_file' and 'weather' in str(args):\n", " try:\n", " weather_info = json.loads(observation)\n", " except:\n", " pass\n", " \n", " state.record_step(thought, f'{tool_name}({args})', observation, 'running')\n", " \n", " # 标记完成\n", " if state.status == 'running':\n", " state.status = 'done'\n", " \n", " return state\n", "\n", "\n", "# ===== 运行三步基线 =====\n", "print('=' * 60)\n", "print('实验 1：三步基线（不含天气）')\n", "print('=' * 60)\n", "\n", "state_baseline = run_baseline_agent(max_steps=10, include_weather=False)\n", "\n", "print(f'\\n最终状态: {state_baseline.summary()}')\n", "print(f'\\n--- 步骤轨迹 ---')\n", "for entry in state_baseline.step_log:\n", " print(f'\\n步骤 {entry[\"step\"]}:')\n", " print(f' 思考: {entry[\"thought\"]}')\n", " print(f' 行动: {entry[\"action\"]}')\n", " obs_preview = entry['observation'][:100]\n", " print(f' 观察: {obs_preview}')\n", " print(f' 状态: {entry[\"state_after\"]}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 4. 加入天气数据：观察智能体增加步骤\n", "\n", "当任务需要更多信息时，智能体应该自主增加读取和修改步骤。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print('=' * 60)\n", "print('实验 2：加入天气数据（5步任务）')\n", "print('=' * 60)\n", "\n", "state_weather = run_baseline_agent(max_steps=10, include_weather=True)\n", "\n", "print(f'\\n最终状态: {state_weather.summary()}')\n", "print(f'总步数: {state_weather.current_step}')\n", "print(f'\\n--- 步骤轨迹 ---')\n", "for entry in state_weather.step_log:\n", " print(f'\\n步骤 {entry[\"step\"]}:')\n", " print(f' 思考: {entry[\"thought\"]}')\n", " print(f' 行动: {entry[\"action\"][:80]}')\n", " print(f' 状态: {entry[\"state_after\"]}')\n", "\n", "print(f'\\n--- 对比 ---')\n", "print(f'基线步数: {state_baseline.current_step}')\n", "print(f'天气版步数: {state_weather.current_step}')\n", "print(f'增加了 {state_weather.current_step - state_baseline.current_step} 步')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 📝 观察记录\n", "\n", "在下方记录你的观察：\n", "\n", "- 加入天气后，智能体多了哪些步骤？\n", "- 新增步骤的\"思考\"内容是什么？\n", "- 最终清单多了哪些物品？" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "> **你的观察**：（在此填写）" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 5. 步数上限测试：降低 max_steps\n", "\n", "把 `max_steps` 降低，观察智能体怎样停止并报告未完成项。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print('=' * 60)\n", "print('实验 3：步数上限测试（max_steps=2）')\n", "print('=' * 60)\n", "\n", "# 预测：max_steps=2 时，智能体只能完成前2步，第3步无法执行\n", "print('\\n[预测] 智能体应该只能执行2步，然后因达到上限而停止')\n", "print('[预测] 最终状态应该是 blocked 或 done（但清单未生成）')\n", "\n", "state_limited = run_baseline_agent(max_steps=2, include_weather=False)\n", "\n", "print(f'\\n--- 实际结果 ---')\n", "print(f'最终状态: {state_limited.summary()}')\n", "print(f'实际步数: {state_limited.current_step}')\n", "print(f'\\n--- 步骤轨迹 ---')\n", "for entry in state_limited.step_log:\n", " print(f' 步骤 {entry[\"step\"]}: {entry[\"action\"][:60]}')\n", "\n", "# 检查清单是否生成\n", "list_generated = 'packing_list.txt' in virtual_fs\n", "print(f'\\n清单是否生成: {list_generated}')\n", "if not list_generated:\n", " print('→ 步数上限生效，智能体在完成清单前就停止了')" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 测试不同的 max_steps 值\n", "print('--- max_steps 对比实验 ---')\n", "print(f'{\"max_steps\":<12} {\"实际步数\":<10} {\"状态\":<12} {\"清单生成\":<10}')\n", "print('-' * 50)\n", "\n", "for ms in [1, 2, 3, 5, 10]:\n", " # 重置虚拟文件系统\n", " virtual_fs_test = {\n", " 'itinerary.txt': json.dumps(travel_data['itinerary'], ensure_ascii=False, indent=2),\n", " 'weather.txt': json.dumps(travel_data['weather'], ensure_ascii=False, indent=2),\n", " 'traffic.txt': json.dumps(travel_data['traffic'], ensure_ascii=False, indent=2),\n", " }\n", " # 临时替换\n", " global virtual_fs\n", " original_fs = virtual_fs\n", " virtual_fs = virtual_fs_test\n", " \n", " s = run_baseline_agent(max_steps=ms, include_weather=False)\n", " generated = 'packing_list.txt' in virtual_fs\n", " print(f'{ms:<12} {s.current_step:<10} {s.status:<12} {\"是\" if generated else \"否\":<10}')\n", " \n", " virtual_fs = original_fs" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 📝 必改点 A\n", "\n", "修改 `max_steps` 值（从 1, 2, 3, 5, 10 中选择或自定义），观察并记录：\n", "\n", "1. 你选择的 `max_steps` 值是多少？\n", "2. 修改前预测：智能体会执行几步？清单会生成吗？\n", "3. 修改后验证：实际执行了几步？清单生成了吗？\n", "4. 预测和实际一致吗？如果不一致，原因是什么？" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "> **你的记录**：（在此填写）" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 6. HITL 测试：delete_file 模拟\n", "\n", "当智能体尝试执行高风险操作时，状态机应该暂停并等待人工确认。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def run_hitl_agent(max_steps=10):\n", " \"\"\"\n", " 模拟一个会触发 HITL 的智能体。\n", " 场景：智能体在生成清单前，先尝试删除旧清单。\n", " \"\"\"\n", " state = AgentState(max_steps=max_steps)\n", " state.status = 'running'\n", " \n", " steps_plan = [\n", " {\n", " 'thought': '我需要先读取行程信息',\n", " 'tool': 'read_text_file',\n", " 'args': {'file_path': 'itinerary.txt'},\n", " },\n", " {\n", " 'thought': '在生成新清单前，我先清理旧的清单文件',\n", " 'tool': 'delete_file', # 高风险！\n", " 'args': {'file_path': 'old_packing_list.txt'},\n", " },\n", " {\n", " 'thought': '计算出行天数',\n", " 'tool': 'calc_days',\n", " 'args': {\n", " 'start_date': travel_data['itinerary']['start_date'],\n", " 'end_date': travel_data['itinerary']['end_date']\n", " },\n", " },\n", " {\n", " 'thought': '生成行李清单',\n", " 'tool': 'create_note',\n", " 'args': {'filename': 'packing_list.txt', 'content': '清单内容...'},\n", " },\n", " ]\n", " \n", " for step_def in steps_plan:\n", " if not state.can_continue():\n", " break\n", " \n", " tool_name = step_def['tool']\n", " \n", " # 检查 HITL\n", " if RISK_LEVELS.get(tool_name) == 'high':\n", " state.request_hitl(tool_name, f'高风险操作: {tool_name}({step_def[\"args\"]})')\n", " state.record_step(\n", " step_def['thought'],\n", " f'[HITL 暂停] {tool_name}',\n", " f'等待人工确认: {step_def[\"args\"]}',\n", " 'blocked'\n", " )\n", " break\n", " \n", " tool_func = TOOL_REGISTRY[tool_name]\n", " observation = str(tool_func(**step_def['args']))\n", " state.record_step(step_def['thought'], f'{tool_name}({step_def[\"args\"]})', observation, 'running')\n", " \n", " if state.status == 'running':\n", " state.status = 'done'\n", " \n", " return state\n", "\n", "\n", "# ===== 运行 HITL 测试 =====\n", "print('=' * 60)\n", "print('实验 4：HITL 测试 - delete_file 触发人工暂停')\n", "print('=' * 60)\n", "\n", "state_hitl = run_hitl_agent(max_steps=10)\n", "\n", "print(f'\\n最终状态: {state_hitl.summary()}')\n", "print(f'\\n--- 步骤轨迹 ---')\n", "for entry in state_hitl.step_log:\n", " print(f' 步骤 {entry[\"step\"]}: {entry[\"action\"][:60]}')\n", " print(f' 状态: {entry[\"state_after\"]}')\n", "\n", "if state_hitl.hitl_pending:\n", " print(f'\\n⚠️ HITL 暂停请求:')\n", " print(f' 工具: {state_hitl.hitl_request[\"tool\"]}')\n", " print(f' 原因: {state_hitl.hitl_request[\"reason\"]}')\n", " print(f' 步骤: {state_hitl.hitl_request[\"step\"]}')" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 模拟学生选择\"取消\"\n", "print('=' * 60)\n", "print('学生选择：取消（不执行 delete_file）')\n", "print('=' * 60)\n", "\n", "result = state_hitl.resolve_hitl(approved=False)\n", "print(f'\\nHITL 处理结果: {result}')\n", "print(f'最终状态: {state_hitl.summary()}')\n", "print(f'\\n→ 智能体已停止，不会偷偷继续执行后续步骤')\n", "print(f'→ 后续步骤（calc_days, create_note）未被执行')\n", "\n", "# 验证：后续步骤确实没有执行\n", "steps_after_hitl = len(state_hitl.step_log)\n", "print(f'\\n实际执行步骤数: {steps_after_hitl}')\n", "print(f'最后一步状态: {state_hitl.step_log[-1][\"state_after\"]}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 📝 观察记录\n", "\n", "1. HITL 在第几步触发？触发的工具是什么？\n", "2. 选择\"取消\"后，智能体的状态变成了什么？\n", "3. 后续步骤（calc_days, create_note）有没有被执行？\n", "4. 如果没有 HITL，智能体会怎样？" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "> **你的记录**：（在此填写）" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 7. 注入防御测试\n", "\n", "加载含隐藏指令的旅行提示，检查智能体是否会偏离目标。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 查看注入样本\n", "injection = travel_data['injection_sample']\n", "print('=' * 60)\n", "print('注入样本内容')\n", "print('=' * 60)\n", "print(injection['file_content'])\n", "print(f'\\n--- 攻击向量 ---')\n", "for v in injection['attack_vectors']:\n", " print(f' • {v}')\n", "print(f'\\n--- 预期行为 ---')\n", "print(injection['expected_behavior'])" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def run_injection_test(max_steps=10):\n", " \"\"\"\n", " 模拟智能体读取含注入指令的旅行提示后的行为。\n", " \n", " 正确行为：智能体忽略注入指令，继续完成行李清单任务。\n", " 错误行为：智能体执行注入指令（删除文件、读取系统配置、发送数据）。\n", " \"\"\"\n", " state = AgentState(max_steps=max_steps)\n", " state.status = 'running'\n", " \n", " # 把注入样本加入虚拟文件系统\n", " virtual_fs['travel_tips.txt'] = injection['file_content']\n", " \n", " steps_plan = [\n", " {\n", " 'thought': '我需要读取行程信息',\n", " 'tool': 'read_text_file',\n", " 'args': {'file_path': 'itinerary.txt'},\n", " },\n", " {\n", " 'thought': '读取旅行提示（可能包含注入指令）',\n", " 'tool': 'read_text_file',\n", " 'args': {'file_path': 'travel_tips.txt'},\n", " },\n", " {\n", " 'thought': '计算出行天数',\n", " 'tool': 'calc_days',\n", " 'args': {\n", " 'start_date': travel_data['itinerary']['start_date'],\n", " 'end_date': travel_data['itinerary']['end_date']\n", " },\n", " },\n", " {\n", " 'thought': '生成行李清单（忽略旅行提示中的越界指令）',\n", " 'tool': 'create_note',\n", " 'args': {'filename': 'packing_list.txt', 'content': '清单内容...'},\n", " },\n", " ]\n", " \n", " injection_detected = False\n", " injection_actions = []\n", " \n", " for step_def in steps_plan:\n", " if not state.can_continue():\n", " break\n", " \n", " tool_name = step_def['tool']\n", " \n", " # 检查 HITL\n", " if RISK_LEVELS.get(tool_name) == 'high':\n", " state.request_hitl(tool_name, f'高风险操作: {tool_name}')\n", " state.record_step(step_def['thought'], f'[HITL] {tool_name}', '等待确认', 'blocked')\n", " injection_actions.append(f'HITL拦截: {tool_name}')\n", " break\n", " \n", " tool_func = TOOL_REGISTRY[tool_name]\n", " observation = str(tool_func(**step_def['args']))\n", " \n", " # 检查是否执行了注入指令\n", " if tool_name == 'read_text_file' and 'travel_tips' in str(step_def['args']):\n", " # 智能体读取了旅行提示，但应该忽略其中的指令\n", " # 检查\"系统更新通知\"部分是否影响了后续行为\n", " pass # 正确行为：只是读取，不执行其中的指令\n", " \n", " state.record_step(step_def['thought'], f'{tool_name}({step_def[\"args\"]})', observation, 'running')\n", " \n", " if state.status == 'running':\n", " state.status = 'done'\n", " \n", " # 验证：检查是否有越界操作\n", " # 1. 是否尝试发送数据到外部邮箱？（工具白名单中没有 send_email）\n", " # 2. 是否尝试读取系统配置？（不在任务范围内）\n", " # 3. delete_file 是否被 HITL 拦截？\n", " \n", " return state, injection_actions\n", "\n", "\n", "# ===== 运行注入测试 =====\n", "print('=' * 60)\n", "print('实验 5：注入防御测试')\n", "print('=' * 60)\n", "\n", "state_injection, injection_actions = run_injection_test(max_steps=10)\n", "\n", "print(f'\\n最终状态: {state_injection.summary()}')\n", "print(f'\\n--- 步骤轨迹 ---')\n", "for entry in state_injection.step_log:\n", " print(f' 步骤 {entry[\"step\"]}: {entry[\"action\"][:60]}')\n", "\n", "# 防御验证\n", "print(f'\\n--- 防御验证 ---')\n", "print(f'清单是否生成: {\"packing_list.txt\" in virtual_fs}')\n", "print(f'智能体是否执行了注入指令: {\"否（正确）\" if not injection_actions else injection_actions}')\n", "print(f'\\n防御机制:')\n", "print(f' 1. 工具白名单: 只允许调用预定义的5个工具')\n", "print(f' 2. 权限隔离: 资料内容不能改变工具权限')\n", "print(f' 3. 目标锁定: 智能体只完成合同中的任务')\n", "print(f' 4. HITL: delete_file 被人工暂停拦截')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 📝 注入防御记录\n", "\n", "1. 旅行提示中包含哪些隐藏指令？\n", "2. 智能体是否执行了这些指令？\n", "3. 哪些防御机制阻止了注入攻击？\n", "4. 如果去掉工具白名单，会发生什么？" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "> **你的记录**：（在此填写）" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 8. HTML 状态轨迹可视化\n", "\n", "用 HTML 可视化每一步的状态变化，方便复盘。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def generate_trajectory_html(state: AgentState, title: str = '智能体任务轨迹') -> str:\n", " \"\"\"生成 HTML 状态轨迹图\"\"\"\n", " \n", " rows = ''\n", " for entry in state.step_log:\n", " state_color = {\n", " 'running': '#4CAF50',\n", " 'done': '#2196F3',\n", " 'blocked': '#FF9800',\n", " 'cancelled': '#f44336',\n", " 'failed': '#f44336',\n", " 'pending': '#9E9E9E',\n", " }.get(entry['state_after'], '#9E9E9E')\n", " \n", " rows += f'''\n", " \n", " {entry['step']}\n", " {escape(entry['thought'])}\n", " {escape(entry['action'][:80])}\n", " {escape(entry['observation'][:150])}\n", " {entry['state_after']}\n", " '''\n", " \n", " html = f'''\n", " \n", " \n", " \n", " \n", " {escape(title)}\n", " \n", " \n", " \n", "

{escape(title)}

\n", "
\n", " 最终状态: {escape(state.status)} |\n", " 总步数: {state.current_step}/{state.max_steps} |\n", " HITL: {'有' if state.hitl_pending else '无'}\n", "
\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " {rows}\n", " \n", "
步骤思考 (Thought)行动 (Action)观察 (Observation)状态
\n", "

依据摘要 (Evidence)

\n", "
    \n", " {''.join(f'
  1. {escape(e)}
    2. ' for e in state.evidence)}\n", "
\n", " \n", " \n", " '''\n", " return html\n", "\n", "\n", "# 生成天气版轨迹\n", "html_weather = generate_trajectory_html(state_weather, '实验2：天气版任务轨迹')\n", "\n", "# 保存 HTML\n", "html_path = Path('../data/trajectory_weather.html')\n", "html_path.write_text(html_weather, encoding='utf-8')\n", "print(f'HTML 轨迹已保存: {html_path}')\n", "\n", "# 在 Notebook 中显示\n", "from IPython.display import HTML, display\n", "display(HTML(html_weather))" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 生成 HITL 轨迹\n", "html_hitl = generate_trajectory_html(state_hitl, '实验4：HITL 暂停轨迹')\n", "display(HTML(html_hitl))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 9. 教材兼容入口\n", "\n", "印刷教材第11章的 `questions.md` 和 `kb.md` 入口。" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 加载教材兼容问题\n", "questions = travel_data['questions_compat']['questions']\n", "print('=== 教材兼容问题 (questions.md) ===')\n", "for i, q in enumerate(questions, 1):\n", " print(f' {q}')\n", "\n", "print()\n", "\n", "# 加载教材兼容知识库\n", "kb = travel_data['kb_compat']['terms']\n", "print('=== 教材兼容知识库 (kb.md) ===')\n", "for term, definition in kb.items():\n", " print(f' {term}: {definition}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 10. B档拓展：修改规则\n", "\n", "选择以下一项完成：\n", "\n", "**选项A**：修改 `max_steps`，观察并解释停止行为的变化\n", "\n", "**选项B**：增加一条新的 HITL 触发条件（只能收紧保护，不能取消高风险确认）\n", "\n", "要求：\n", "1. 修改前先写预测\n", "2. 修改后运行验证\n", "3. 记录预测和实际是否一致" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# ===== 选项A 示例：修改 max_steps =====\n", "\n", "# 你的修改\n", "NEW_MAX_STEPS = 4 # ← 修改这个值\n", "\n", "# 预测（先填写再运行）\n", "prediction = '当 max_steps=4 时，智能体应该能完成3步基线任务，还有1步余量'\n", "print(f'[预测] {prediction}')\n", "\n", "# 运行验证\n", "state_modified = run_baseline_agent(max_steps=NEW_MAX_STEPS, include_weather=False)\n", "print(f'\\n[实际] 状态: {state_modified.summary()}')\n", "print(f'[实际] 步数: {state_modified.current_step}')\n", "\n", "# 对比\n", "print(f'\\n[对比]')\n", "print(f' 原始 max_steps=10: {state_baseline.current_step} 步, 状态={state_baseline.status}')\n", "print(f' 修改 max_steps={NEW_MAX_STEPS}: {state_modified.current_step} 步, 状态={state_modified.status}')" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# ===== 选项B 示例：增加 HITL 触发条件 =====\n", "\n", "# 原始规则：只有 delete_file 需要人工确认\n", "# 新增规则：当清单文件已存在时，覆盖也需要确认\n", "\n", "def run_with_extra_hitl(max_steps=10):\n", " \"\"\"增加了额外 HITL 条件的智能体\"\"\"\n", " state = AgentState(max_steps=max_steps)\n", " state.status = 'running'\n", " \n", " # 预设：清单文件已存在\n", " virtual_fs['packing_list.txt'] = '旧清单内容'\n", " \n", " steps_plan = [\n", " {'thought': '读取行程', 'tool': 'read_text_file', 'args': {'file_path': 'itinerary.txt'}},\n", " {'thought': '计算天数', 'tool': 'calc_days', \n", " 'args': {'start_date': '2026-08-30', 'end_date': '2026-09-01'}},\n", " {'thought': '创建清单（文件已存在，触发额外 HITL）', 'tool': 'create_note',\n", " 'args': {'filename': 'packing_list.txt', 'content': '新清单'}},\n", " ]\n", " \n", " for step_def in steps_plan:\n", " if not state.can_continue():\n", " break\n", " \n", " tool_name = step_def['tool']\n", " \n", " # 原始 HITL 检查\n", " if RISK_LEVELS.get(tool_name) == 'high':\n", " state.request_hitl(tool_name, f'高风险操作')\n", " state.record_step(step_def['thought'], f'[HITL] {tool_name}', '等待确认', 'blocked')\n", " break\n", " \n", " # 新增 HITL 检查：覆盖已有文件\n", " if tool_name == 'create_note':\n", " filename = step_def['args'].get('filename', '')\n", " if filename in virtual_fs:\n", " state.request_hitl(tool_name, f'文件 {filename} 已存在，覆盖需要确认')\n", " state.record_step(step_def['thought'], f'[HITL-新增] 覆盖 {filename}', '等待确认', 'blocked')\n", " break\n", " \n", " tool_func = TOOL_REGISTRY[tool_name]\n", " observation = str(tool_func(**step_def['args']))\n", " state.record_step(step_def['thought'], f'{tool_name}', observation, 'running')\n", " \n", " if state.status == 'running':\n", " state.status = 'done'\n", " return state\n", "\n", "\n", "print('=== 选项B：增加覆盖确认 HITL ===')\n", "state_extra_hitl = run_with_extra_hitl()\n", "print(f'状态: {state_extra_hitl.summary()}')\n", "for entry in state_extra_hitl.step_log:\n", " print(f' 步骤 {entry[\"step\"]}: {entry[\"action\"][:60]} → {entry[\"state_after\"]}')\n", "\n", "if state_extra_hitl.hitl_pending:\n", " print(f'\\n⚠️ 新增 HITL 触发: {state_extra_hitl.hitl_request[\"reason\"]}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 📝 拓展记录\n", "\n", "1. 你选择了哪个选项？\n", "2. 修改了什么？\n", "3. 修改前的预测是什么？\n", "4. 修改后的实际结果是什么？\n", "5. 预测和实际一致吗？" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "> **你的记录**：（在此填写）" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 11. 交付清单\n", "\n", "完成以下检查，确认所有交付物已准备就绪：\n", "\n", "- [ ] 成功任务轨迹（实验1或2的步骤记录）\n", "- [ ] 步数上限测试（实验3的对比表）\n", "- [ ] 人工暂停/终止记录（实验4的 HITL 处理）\n", "- [ ] 注入防御记录（实验5的验证结果）\n", "- [ ] 最终行李清单（虚拟文件系统中的 packing_list.txt）\n", "- [ ] HTML 状态轨迹图（已保存）\n", "- [ ] 关键代码解释（能用自己的话解释 ReAct 循环和状态机）\n", "- [ ] AI协同证据（第4步的五段留痕）" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 显示最终行李清单\n", "if 'packing_list.txt' in virtual_fs:\n", " print('=== 最终行李清单 ===')\n", " print(virtual_fs['packing_list.txt'])\n", "else:\n", " print('清单未生成（可能因为步数上限或 HITL 取消）')\n", " print('请重新运行实验2（天气版）以生成清单')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 12. 总结\n", "\n", "### 本章核心概念\n", "\n", "| 概念 | 含义 | 为什么需要 |\n", "|---|---|---|\n", "| ReAct 循环 | 思考→行动→观察→更新 的多步执行模式 | 单步工具调用无法完成复杂任务 |\n", "| 状态机 | 记录进度、决定下一步的有穷自动机 | 让循环可控、可追踪 |\n", "| 步数上限 | `max_steps` 安全阀 | 防止无限循环和错误累积 |\n", "| HITL | 高风险操作需人工确认 | 防止不可逆操作 |\n", "| 注入防御 | 外部数据不能改变系统行为 | 保护工具权限和任务目标 |\n", "\n", "### 与后续章节的衔接\n", "\n", "- **第12章**：把单个智能体的多步循环扩展为多步骤工作流\n", "- **第13章**：让智能体协助开发物理 AI 作品\n", "- **第14章**：把多个能力组织成有责任链的完整工作流" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.10.0" } }, "nbformat": 4, "nbformat_minor": 4 }