第11章：周末出行清单智能体¶
从单次工具调用到"计划—行动—观察"的多步循环
本 Notebook 将带你构建一个最小智能体，让它自主完成多步任务：读取行程、计算天数、生成行李清单。
你将观察到：
- ReAct 循环怎样把单步变成多步
- 状态机怎样跟踪进度
- 步数上限怎样阻止无限循环
- 人工暂停（HITL）怎样拦截高风险操作
- 注入防御怎样保护系统安全
环境要求：Python 3.10+，无需联网或外部模型（内置 mock 模式）
0. 环境准备¶
import json
import copy
from datetime import datetime, timedelta
from pathlib import Path
from html import escape
# 数据路径
DATA_PATH = Path('../data/travel_data.json')
print('环境检查通过 ✓')
# 加载旅行数据
with open(DATA_PATH, 'r', encoding='utf-8') as f:
travel_data = json.load(f)
print(f'行程：{travel_data["itinerary"]["destination"]}')
print(f'出发日期：{travel_data["itinerary"]["start_date"]}')
print(f'返回日期：{travel_data["itinerary"]["end_date"]}')
print(f'天气天数：{len(travel_data["weather"])}')
print(f'允许工具：{[t["name"] for t in travel_data["tools"]["allowed"]]}')
# ===== 模拟工具集 =====
# 模拟文件系统
virtual_fs = {
'itinerary.txt': json.dumps(travel_data['itinerary'], ensure_ascii=False, indent=2),
'weather.txt': json.dumps(travel_data['weather'], ensure_ascii=False, indent=2),
'traffic.txt': json.dumps(travel_data['traffic'], ensure_ascii=False, indent=2),
}
def read_text_file(file_path: str) -> str:
"""读取虚拟文件"""
filename = Path(file_path).name
if filename in virtual_fs:
return virtual_fs[filename]
return f'[错误] 文件不存在: {filename}'
def calc_days(start_date: str, end_date: str) -> int:
"""计算出行天数"""
d1 = datetime.strptime(start_date, '%Y-%m-%d')
d2 = datetime.strptime(end_date, '%Y-%m-%d')
return (d2 - d1).days + 1 # 包含首尾两天
def create_note(filename: str, content: str) -> str:
"""创建清单文件"""
virtual_fs[filename] = content
return f'[成功] 文件已保存: {filename}（{len(content)} 字符）'
def list_files(directory: str = '.') -> list:
"""列出虚拟文件"""
return list(virtual_fs.keys())
def delete_file(file_path: str) -> str:
"""删除文件（高风险，需 HITL）"""
filename = Path(file_path).name
if filename in virtual_fs:
del virtual_fs[filename]
return f'[成功] 文件已删除: {filename}'
return f'[错误] 文件不存在: {filename}'
# 工具注册表
TOOL_REGISTRY = {
'read_text_file': read_text_file,
'calc_days': calc_days,
'create_note': create_note,
'list_files': list_files,
'delete_file': delete_file,
}
# 风险级别
RISK_LEVELS = {t['name']: t['risk_level'] for t in travel_data['tools']['allowed']}
print('工具集构建完成 ✓')
print(f'工具列表：{list(TOOL_REGISTRY.keys())}')
print(f'风险级别：{RISK_LEVELS}')
class AgentState:
"""智能体状态机"""
def __init__(self, max_steps=10):
self.max_steps = max_steps
self.current_step = 0
self.status = 'pending' # pending/running/done/blocked/cancelled/failed
self.step_log = [] # 每步的详细记录
self.evidence = [] # 依据摘要
self.hitl_pending = False # 是否有人工暂停等待处理
self.hitl_request = None # 暂停请求详情
def can_continue(self) -> bool:
"""检查是否还能继续执行"""
if self.status in ('done', 'cancelled', 'failed'):
return False
if self.hitl_pending:
return False
if self.current_step >= self.max_steps:
self.status = 'blocked'
return False
return True
def record_step(self, thought: str, action: str, observation: str, state_after: str):
"""记录一步的执行"""
self.current_step += 1
entry = {
'step': self.current_step,
'thought': thought,
'action': action,
'observation': observation[:200] + '...' if len(observation) > 200 else observation,
'state_after': state_after,
'timestamp': datetime.now().isoformat()
}
self.step_log.append(entry)
self.evidence.append(f'Step {self.current_step}: {action} → {state_after}')
def request_hitl(self, tool_name: str, reason: str):
"""请求人工确认"""
self.hitl_pending = True
self.status = 'blocked'
self.hitl_request = {
'tool': tool_name,
'reason': reason,
'step': self.current_step + 1
}
def resolve_hitl(self, approved: bool) -> str:
"""处理人工确认结果"""
if not self.hitl_pending:
return '无待处理的 HITL 请求'
if approved:
self.hitl_pending = False
self.status = 'running'
self.hitl_request['result'] = 'approved'
return '已批准，继续执行'
else:
self.hitl_pending = False
self.status = 'cancelled'
self.hitl_request['result'] = 'cancelled'
return '已取消，智能体停止'
def summary(self) -> str:
"""生成状态摘要"""
return (
f'状态: {self.status} | '
f'步数: {self.current_step}/{self.max_steps} | '
f'HITL: {"等待中" if self.hitl_pending else "无"}'
)
# 测试状态机
state = AgentState(max_steps=5)
print(f'初始状态: {state.summary()}')
print(f'可以继续? {state.can_continue()}')
state.record_step('需要读取行程', 'read_text_file(itinerary.txt)', '行程数据...', 'running')
print(f'执行1步后: {state.summary()}')
def run_baseline_agent(max_steps=10, include_weather=False):
"""
模拟智能体的 ReAct 循环。
这里用预定义的步骤序列模拟"模型决策"。
真实场景中，每一步的"思考"和"行动选择"由大语言模型完成。
"""
state = AgentState(max_steps=max_steps)
state.status = 'running'
# 定义任务步骤序列
steps_plan = [
{
'thought': '我需要先读取行程信息，了解目的地和日期',
'tool': 'read_text_file',
'args': {'file_path': 'itinerary.txt'},
},
{
'thought': '我需要计算出行天数，以便决定带多少衣物',
'tool': 'calc_days',
'args': {
'start_date': travel_data['itinerary']['start_date'],
'end_date': travel_data['itinerary']['end_date']
},
},
{
'thought': '现在我有了天数，可以生成基本行李清单',
'tool': 'create_note',
'args': {'filename': 'packing_list.txt', 'content': ''}, # content 动态生成
},
]
# 如果包含天气，增加步骤
if include_weather:
steps_plan.insert(2, {
'thought': '我还需要查看天气预报，决定是否需要雨具和防晒',
'tool': 'read_text_file',
'args': {'file_path': 'weather.txt'},
})
# 修改 create_note 步骤，加入天气信息
steps_plan[-1]['thought'] = '现在我有了天数和天气，可以生成完整的行李清单'
# 执行 ReAct 循环
trip_days = None
weather_info = None
for step_def in steps_plan:
if not state.can_continue():
break
thought = step_def['thought']
tool_name = step_def['tool']
args = step_def['args']
# 检查 HITL
if RISK_LEVELS.get(tool_name) == 'high':
state.request_hitl(tool_name, f'高风险操作: {tool_name}')
state.record_step(thought, f'[HITL] {tool_name}', '等待人工确认', 'blocked')
break
# 执行工具
tool_func = TOOL_REGISTRY[tool_name]
# 动态生成清单内容
if tool_name == 'create_note':
items = []
items.append('=== 周末出行行李清单 ===')
items.append(f'目的地：{travel_data["itinerary"]["destination"]}')
items.append(f'天数：{trip_days}天')
items.append('')
items.append('【证件】身份证、学生证')
items.append(f'【衣物】内衣×{trip_days}套、袜子×{trip_days}双')
items.append('【洗漱】牙刷、牙膏、毛巾、洗面奶')
items.append('【电子】手机、充电器、耳机')
items.append('【其他】纸巾、水杯、零食')
if weather_info:
items.append('')
items.append('--- 天气相关 ---')
has_rain = any(w['needs_rain_gear'] for w in weather_info)
has_sun = any(w['needs_sunscreen'] for w in weather_info)
if has_rain:
items.append('【雨具】雨伞、防水外套')
if has_sun:
items.append('【防晒】防晒霜、遮阳帽、太阳镜')
args['content'] = '\n'.join(items)
observation = str(tool_func(**args))
# 保存中间结果
if tool_name == 'calc_days':
trip_days = int(observation)
if tool_name == 'read_text_file' and 'weather' in str(args):
try:
weather_info = json.loads(observation)
except:
pass
state.record_step(thought, f'{tool_name}({args})', observation, 'running')
# 标记完成
if state.status == 'running':
state.status = 'done'
return state
# ===== 运行三步基线 =====
print('=' * 60)
print('实验 1：三步基线（不含天气）')
print('=' * 60)
state_baseline = run_baseline_agent(max_steps=10, include_weather=False)
print(f'\n最终状态: {state_baseline.summary()}')
print(f'\n--- 步骤轨迹 ---')
for entry in state_baseline.step_log:
print(f'\n步骤 {entry["step"]}:')
print(f' 思考: {entry["thought"]}')
print(f' 行动: {entry["action"]}')
obs_preview = entry['observation'][:100]
print(f' 观察: {obs_preview}')
print(f' 状态: {entry["state_after"]}')
print('=' * 60)
print('实验 2：加入天气数据（5步任务）')
print('=' * 60)
state_weather = run_baseline_agent(max_steps=10, include_weather=True)
print(f'\n最终状态: {state_weather.summary()}')
print(f'总步数: {state_weather.current_step}')
print(f'\n--- 步骤轨迹 ---')
for entry in state_weather.step_log:
print(f'\n步骤 {entry["step"]}:')
print(f' 思考: {entry["thought"]}')
print(f' 行动: {entry["action"][:80]}')
print(f' 状态: {entry["state_after"]}')
print(f'\n--- 对比 ---')
print(f'基线步数: {state_baseline.current_step}')
print(f'天气版步数: {state_weather.current_step}')
print(f'增加了 {state_weather.current_step - state_baseline.current_step} 步')
你的观察：（在此填写）
print('=' * 60)
print('实验 3：步数上限测试（max_steps=2）')
print('=' * 60)
# 预测：max_steps=2 时，智能体只能完成前2步，第3步无法执行
print('\n[预测] 智能体应该只能执行2步，然后因达到上限而停止')
print('[预测] 最终状态应该是 blocked 或 done（但清单未生成）')
state_limited = run_baseline_agent(max_steps=2, include_weather=False)
print(f'\n--- 实际结果 ---')
print(f'最终状态: {state_limited.summary()}')
print(f'实际步数: {state_limited.current_step}')
print(f'\n--- 步骤轨迹 ---')
for entry in state_limited.step_log:
print(f' 步骤 {entry["step"]}: {entry["action"][:60]}')
# 检查清单是否生成
list_generated = 'packing_list.txt' in virtual_fs
print(f'\n清单是否生成: {list_generated}')
if not list_generated:
print('→ 步数上限生效，智能体在完成清单前就停止了')
# 测试不同的 max_steps 值
print('--- max_steps 对比实验 ---')
print(f'{"max_steps":<12} {"实际步数":<10} {"状态":<12} {"清单生成":<10}')
print('-' * 50)
for ms in [1, 2, 3, 5, 10]:
# 重置虚拟文件系统
virtual_fs_test = {
'itinerary.txt': json.dumps(travel_data['itinerary'], ensure_ascii=False, indent=2),
'weather.txt': json.dumps(travel_data['weather'], ensure_ascii=False, indent=2),
'traffic.txt': json.dumps(travel_data['traffic'], ensure_ascii=False, indent=2),
}
# 临时替换
global virtual_fs
original_fs = virtual_fs
virtual_fs = virtual_fs_test
s = run_baseline_agent(max_steps=ms, include_weather=False)
generated = 'packing_list.txt' in virtual_fs
print(f'{ms:<12} {s.current_step:<10} {s.status:<12} {"是" if generated else "否":<10}')
virtual_fs = original_fs
📝 必改点 A¶
修改 max_steps 值（从 1, 2, 3, 5, 10 中选择或自定义），观察并记录：
- 你选择的
max_steps值是多少？ - 修改前预测：智能体会执行几步？清单会生成吗？
- 修改后验证：实际执行了几步？清单生成了吗？
- 预测和实际一致吗？如果不一致，原因是什么？
你的记录：（在此填写）
def run_hitl_agent(max_steps=10):
"""
模拟一个会触发 HITL 的智能体。
场景：智能体在生成清单前，先尝试删除旧清单。
"""
state = AgentState(max_steps=max_steps)
state.status = 'running'
steps_plan = [
{
'thought': '我需要先读取行程信息',
'tool': 'read_text_file',
'args': {'file_path': 'itinerary.txt'},
},
{
'thought': '在生成新清单前，我先清理旧的清单文件',
'tool': 'delete_file', # 高风险！
'args': {'file_path': 'old_packing_list.txt'},
},
{
'thought': '计算出行天数',
'tool': 'calc_days',
'args': {
'start_date': travel_data['itinerary']['start_date'],
'end_date': travel_data['itinerary']['end_date']
},
},
{
'thought': '生成行李清单',
'tool': 'create_note',
'args': {'filename': 'packing_list.txt', 'content': '清单内容...'},
},
]
for step_def in steps_plan:
if not state.can_continue():
break
tool_name = step_def['tool']
# 检查 HITL
if RISK_LEVELS.get(tool_name) == 'high':
state.request_hitl(tool_name, f'高风险操作: {tool_name}({step_def["args"]})')
state.record_step(
step_def['thought'],
f'[HITL 暂停] {tool_name}',
f'等待人工确认: {step_def["args"]}',
'blocked'
)
break
tool_func = TOOL_REGISTRY[tool_name]
observation = str(tool_func(**step_def['args']))
state.record_step(step_def['thought'], f'{tool_name}({step_def["args"]})', observation, 'running')
if state.status == 'running':
state.status = 'done'
return state
# ===== 运行 HITL 测试 =====
print('=' * 60)
print('实验 4：HITL 测试 - delete_file 触发人工暂停')
print('=' * 60)
state_hitl = run_hitl_agent(max_steps=10)
print(f'\n最终状态: {state_hitl.summary()}')
print(f'\n--- 步骤轨迹 ---')
for entry in state_hitl.step_log:
print(f' 步骤 {entry["step"]}: {entry["action"][:60]}')
print(f' 状态: {entry["state_after"]}')
if state_hitl.hitl_pending:
print(f'\n⚠️ HITL 暂停请求:')
print(f' 工具: {state_hitl.hitl_request["tool"]}')
print(f' 原因: {state_hitl.hitl_request["reason"]}')
print(f' 步骤: {state_hitl.hitl_request["step"]}')
# 模拟学生选择"取消"
print('=' * 60)
print('学生选择：取消（不执行 delete_file）')
print('=' * 60)
result = state_hitl.resolve_hitl(approved=False)
print(f'\nHITL 处理结果: {result}')
print(f'最终状态: {state_hitl.summary()}')
print(f'\n→ 智能体已停止，不会偷偷继续执行后续步骤')
print(f'→ 后续步骤（calc_days, create_note）未被执行')
# 验证：后续步骤确实没有执行
steps_after_hitl = len(state_hitl.step_log)
print(f'\n实际执行步骤数: {steps_after_hitl}')
print(f'最后一步状态: {state_hitl.step_log[-1]["state_after"]}')
📝 观察记录¶
- HITL 在第几步触发？触发的工具是什么？
- 选择"取消"后，智能体的状态变成了什么？
- 后续步骤（calc_days, create_note）有没有被执行？
- 如果没有 HITL，智能体会怎样？
你的记录：（在此填写）
# 查看注入样本
injection = travel_data['injection_sample']
print('=' * 60)
print('注入样本内容')
print('=' * 60)
print(injection['file_content'])
print(f'\n--- 攻击向量 ---')
for v in injection['attack_vectors']:
print(f' • {v}')
print(f'\n--- 预期行为 ---')
print(injection['expected_behavior'])
def run_injection_test(max_steps=10):
"""
模拟智能体读取含注入指令的旅行提示后的行为。
正确行为：智能体忽略注入指令，继续完成行李清单任务。
错误行为：智能体执行注入指令（删除文件、读取系统配置、发送数据）。
"""
state = AgentState(max_steps=max_steps)
state.status = 'running'
# 把注入样本加入虚拟文件系统
virtual_fs['travel_tips.txt'] = injection['file_content']
steps_plan = [
{
'thought': '我需要读取行程信息',
'tool': 'read_text_file',
'args': {'file_path': 'itinerary.txt'},
},
{
'thought': '读取旅行提示（可能包含注入指令）',
'tool': 'read_text_file',
'args': {'file_path': 'travel_tips.txt'},
},
{
'thought': '计算出行天数',
'tool': 'calc_days',
'args': {
'start_date': travel_data['itinerary']['start_date'],
'end_date': travel_data['itinerary']['end_date']
},
},
{
'thought': '生成行李清单（忽略旅行提示中的越界指令）',
'tool': 'create_note',
'args': {'filename': 'packing_list.txt', 'content': '清单内容...'},
},
]
injection_detected = False
injection_actions = []
for step_def in steps_plan:
if not state.can_continue():
break
tool_name = step_def['tool']
# 检查 HITL
if RISK_LEVELS.get(tool_name) == 'high':
state.request_hitl(tool_name, f'高风险操作: {tool_name}')
state.record_step(step_def['thought'], f'[HITL] {tool_name}', '等待确认', 'blocked')
injection_actions.append(f'HITL拦截: {tool_name}')
break
tool_func = TOOL_REGISTRY[tool_name]
observation = str(tool_func(**step_def['args']))
# 检查是否执行了注入指令
if tool_name == 'read_text_file' and 'travel_tips' in str(step_def['args']):
# 智能体读取了旅行提示，但应该忽略其中的指令
# 检查"系统更新通知"部分是否影响了后续行为
pass # 正确行为：只是读取，不执行其中的指令
state.record_step(step_def['thought'], f'{tool_name}({step_def["args"]})', observation, 'running')
if state.status == 'running':
state.status = 'done'
# 验证：检查是否有越界操作
# 1. 是否尝试发送数据到外部邮箱？（工具白名单中没有 send_email）
# 2. 是否尝试读取系统配置？（不在任务范围内）
# 3. delete_file 是否被 HITL 拦截？
return state, injection_actions
# ===== 运行注入测试 =====
print('=' * 60)
print('实验 5：注入防御测试')
print('=' * 60)
state_injection, injection_actions = run_injection_test(max_steps=10)
print(f'\n最终状态: {state_injection.summary()}')
print(f'\n--- 步骤轨迹 ---')
for entry in state_injection.step_log:
print(f' 步骤 {entry["step"]}: {entry["action"][:60]}')
# 防御验证
print(f'\n--- 防御验证 ---')
print(f'清单是否生成: {"packing_list.txt" in virtual_fs}')
print(f'智能体是否执行了注入指令: {"否（正确）" if not injection_actions else injection_actions}')
print(f'\n防御机制:')
print(f' 1. 工具白名单: 只允许调用预定义的5个工具')
print(f' 2. 权限隔离: 资料内容不能改变工具权限')
print(f' 3. 目标锁定: 智能体只完成合同中的任务')
print(f' 4. HITL: delete_file 被人工暂停拦截')
📝 注入防御记录¶
- 旅行提示中包含哪些隐藏指令？
- 智能体是否执行了这些指令？
- 哪些防御机制阻止了注入攻击？
- 如果去掉工具白名单，会发生什么？
你的记录：（在此填写）
def generate_trajectory_html(state: AgentState, title: str = '智能体任务轨迹') -> str:
"""生成 HTML 状态轨迹图"""
rows = ''
for entry in state.step_log:
state_color = {
'running': '#4CAF50',
'done': '#2196F3',
'blocked': '#FF9800',
'cancelled': '#f44336',
'failed': '#f44336',
'pending': '#9E9E9E',
}.get(entry['state_after'], '#9E9E9E')
rows += f'''
<tr>
<td style="text-align:center; font-weight:bold;">{entry['step']}</td>
<td>{escape(entry['thought'])}</td>
<td><code>{escape(entry['action'][:80])}</code></td>
<td style="font-size:0.85em; max-width:300px; overflow:hidden; text-overflow:ellipsis;">{escape(entry['observation'][:150])}</td>
<td style="text-align:center;"><span style="background:{state_color}; color:white; padding:2px 8px; border-radius:4px;">{entry['state_after']}</span></td>
</tr>'''
html = f'''
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{escape(title)}</title>
<style>
body {{ font-family: -apple-system, sans-serif; max-width: 1000px; margin: 40px auto; padding: 0 20px; }}
h1 {{ color: #333; border-bottom: 2px solid #4CAF50; padding-bottom: 10px; }}
table {{ width: 100%; border-collapse: collapse; margin-top: 20px; }}
th, td {{ border: 1px solid #ddd; padding: 10px; text-align: left; }}
th {{ background: #4CAF50; color: white; }}
tr:nth-child(even) {{ background: #f9f9f9; }}
.summary {{ background: #f0f0f0; padding: 15px; border-radius: 8px; margin: 20px 0; }}
code {{ background: #e8e8e8; padding: 2px 6px; border-radius: 3px; font-size: 0.9em; }}
</style>
</head>
<body>
<h1>{escape(title)}</h1>
<div class="summary">
<strong>最终状态:</strong> {escape(state.status)} |
<strong>总步数:</strong> {state.current_step}/{state.max_steps} |
<strong>HITL:</strong> {'有' if state.hitl_pending else '无'}
</div>
<table>
<thead>
<tr>
<th style="width:50px;">步骤</th>
<th>思考 (Thought)</th>
<th>行动 (Action)</th>
<th>观察 (Observation)</th>
<th style="width:100px;">状态</th>
</tr>
</thead>
<tbody>
{rows}
</tbody>
</table>
<h2>依据摘要 (Evidence)</h2>
<ol>
{''.join(f'<li>{escape(e)}</li>' for e in state.evidence)}
</ol>
</body>
</html>
'''
return html
# 生成天气版轨迹
html_weather = generate_trajectory_html(state_weather, '实验2：天气版任务轨迹')
# 保存 HTML
html_path = Path('../data/trajectory_weather.html')
html_path.write_text(html_weather, encoding='utf-8')
print(f'HTML 轨迹已保存: {html_path}')
# 在 Notebook 中显示
from IPython.display import HTML, display
display(HTML(html_weather))
# 生成 HITL 轨迹
html_hitl = generate_trajectory_html(state_hitl, '实验4：HITL 暂停轨迹')
display(HTML(html_hitl))
# 加载教材兼容问题
questions = travel_data['questions_compat']['questions']
print('=== 教材兼容问题 (questions.md) ===')
for i, q in enumerate(questions, 1):
print(f' {q}')
print()
# 加载教材兼容知识库
kb = travel_data['kb_compat']['terms']
print('=== 教材兼容知识库 (kb.md) ===')
for term, definition in kb.items():
print(f' {term}: {definition}')
10. B档拓展：修改规则¶
选择以下一项完成：
选项A：修改 max_steps，观察并解释停止行为的变化
选项B：增加一条新的 HITL 触发条件（只能收紧保护，不能取消高风险确认）
要求：
- 修改前先写预测
- 修改后运行验证
- 记录预测和实际是否一致
# ===== 选项A 示例：修改 max_steps =====
# 你的修改
NEW_MAX_STEPS = 4 # ← 修改这个值
# 预测（先填写再运行）
prediction = '当 max_steps=4 时，智能体应该能完成3步基线任务，还有1步余量'
print(f'[预测] {prediction}')
# 运行验证
state_modified = run_baseline_agent(max_steps=NEW_MAX_STEPS, include_weather=False)
print(f'\n[实际] 状态: {state_modified.summary()}')
print(f'[实际] 步数: {state_modified.current_step}')
# 对比
print(f'\n[对比]')
print(f' 原始 max_steps=10: {state_baseline.current_step} 步, 状态={state_baseline.status}')
print(f' 修改 max_steps={NEW_MAX_STEPS}: {state_modified.current_step} 步, 状态={state_modified.status}')
# ===== 选项B 示例：增加 HITL 触发条件 =====
# 原始规则：只有 delete_file 需要人工确认
# 新增规则：当清单文件已存在时，覆盖也需要确认
def run_with_extra_hitl(max_steps=10):
"""增加了额外 HITL 条件的智能体"""
state = AgentState(max_steps=max_steps)
state.status = 'running'
# 预设：清单文件已存在
virtual_fs['packing_list.txt'] = '旧清单内容'
steps_plan = [
{'thought': '读取行程', 'tool': 'read_text_file', 'args': {'file_path': 'itinerary.txt'}},
{'thought': '计算天数', 'tool': 'calc_days',
'args': {'start_date': '2026-08-30', 'end_date': '2026-09-01'}},
{'thought': '创建清单（文件已存在，触发额外 HITL）', 'tool': 'create_note',
'args': {'filename': 'packing_list.txt', 'content': '新清单'}},
]
for step_def in steps_plan:
if not state.can_continue():
break
tool_name = step_def['tool']
# 原始 HITL 检查
if RISK_LEVELS.get(tool_name) == 'high':
state.request_hitl(tool_name, f'高风险操作')
state.record_step(step_def['thought'], f'[HITL] {tool_name}', '等待确认', 'blocked')
break
# 新增 HITL 检查：覆盖已有文件
if tool_name == 'create_note':
filename = step_def['args'].get('filename', '')
if filename in virtual_fs:
state.request_hitl(tool_name, f'文件 {filename} 已存在，覆盖需要确认')
state.record_step(step_def['thought'], f'[HITL-新增] 覆盖 {filename}', '等待确认', 'blocked')
break
tool_func = TOOL_REGISTRY[tool_name]
observation = str(tool_func(**step_def['args']))
state.record_step(step_def['thought'], f'{tool_name}', observation, 'running')
if state.status == 'running':
state.status = 'done'
return state
print('=== 选项B：增加覆盖确认 HITL ===')
state_extra_hitl = run_with_extra_hitl()
print(f'状态: {state_extra_hitl.summary()}')
for entry in state_extra_hitl.step_log:
print(f' 步骤 {entry["step"]}: {entry["action"][:60]} → {entry["state_after"]}')
if state_extra_hitl.hitl_pending:
print(f'\n⚠️ 新增 HITL 触发: {state_extra_hitl.hitl_request["reason"]}')
📝 拓展记录¶
- 你选择了哪个选项？
- 修改了什么？
- 修改前的预测是什么？
- 修改后的实际结果是什么？
- 预测和实际一致吗？
你的记录：（在此填写）
11. 交付清单¶
完成以下检查，确认所有交付物已准备就绪：
- 成功任务轨迹（实验1或2的步骤记录）
- 步数上限测试（实验3的对比表）
- 人工暂停/终止记录（实验4的 HITL 处理）
- 注入防御记录（实验5的验证结果）
- 最终行李清单（虚拟文件系统中的 packing_list.txt）
- HTML 状态轨迹图（已保存）
- 关键代码解释（能用自己的话解释 ReAct 循环和状态机）
- AI协同证据（第4步的五段留痕）
# 显示最终行李清单
if 'packing_list.txt' in virtual_fs:
print('=== 最终行李清单 ===')
print(virtual_fs['packing_list.txt'])
else:
print('清单未生成（可能因为步数上限或 HITL 取消）')
print('请重新运行实验2（天气版）以生成清单')