第 24 章
Files API：PDF/图片跨请求复用、500MB 文件处理与 ZDR 限制
第二十四章：Advisor Tool：Claude 的元认知与任务规划能力
24.1 元认知在 AI Agent 中的价值
元认知（Metacognition）是"对认知的认知"——即能够反思自己的思维过程、评估自己的能力边界、规划执行策略的能力。对于 AI Agent 来说，元认知能力意味着在执行任务之前，能够回答以下问题：
- 我是否充分理解了这个任务？
- 完成这个任务需要哪些步骤？
- 我是否具备完成每个步骤的能力？
- 哪些步骤风险最高，需要额外谨慎？
- 如果某个步骤失败，该如何恢复？
Advisor Tool 是一种设计模式，它将 Claude 的元认知能力显式化为可观测、可审计的工具调用，使得任务规划过程透明且可控。
24.2 Advisor Tool 的设计哲学
Advisor Tool 的核心思想是：在 Claude 开始行动之前，先让它作为"顾问"审查整个任务。
这种模式的灵感来自人类的"双系统"决策理论：
- 系统一（快速、直觉）：立即开始执行，凭经验快速响应
- 系统二（慢速、分析）：先停下来，系统地分析再行动
对于复杂的 Agent 任务，直接进入"系统一"执行模式往往会导致：
- 中途发现缺少关键信息
- 错误的工具调用顺序
- 忽略边界情况导致数据损坏
- 无法恢复的操作（如发送了测试邮件给真实用户）
Advisor Tool 强制 Claude 先进入"系统二"模式。
24.3 Advisor Tool 的完整定义
advisor_tool = {
"name": "advisor",
"description": """在执行任何复杂任务之前，调用此工具进行元认知评估和任务规划。
你应该在以下情况使用此工具：
1. 任务涉及多个步骤或多个工具的组合
2. 任务涉及不可逆操作（删除、发送、提交等）
3. 任务需求不够明确，需要澄清
4. 任务失败的代价较高
此工具不执行任何实际操作，只进行规划和评估。
【Advisor 评估输出包括】
- task_understanding: 对任务的理解与潜在歧义
- execution_plan: 详细的执行步骤（含顺序和依赖关系）
- risk_assessment: 每个步骤的风险等级和缓解措施
- capability_check: 完成任务所需的工具和信息清单
- clarifications_needed: 需要用户澄清的问题（如有）
- recommendation: proceed（继续）/ clarify（先澄清）/ reject（拒绝执行）""",
"input_schema": {
"type": "object",
"properties": {
"task_description": {
"type": "string",
"description": "用你自己的话描述你理解的任务（不是重复用户原文）"
},
"task_complexity": {
"type": "string",
"enum": ["simple", "moderate", "complex", "critical"],
"description": "任务复杂度评估：simple(单步骤) / moderate(多步骤) / complex(多工具多步骤) / critical(高风险或不可逆)"
},
"proposed_steps": {
"type": "array",
"description": "计划执行的步骤列表",
"items": {
"type": "object",
"properties": {
"step_number": {"type": "integer"},
"action": {"type": "string", "description": "步骤描述"},
"tool_required": {"type": "string", "description": "需要的工具名称"},
"dependencies": {
"type": "array",
"items": {"type": "integer"},
"description": "依赖的前置步骤编号"
},
"risk_level": {
"type": "string",
"enum": ["low", "medium", "high", "critical"],
"description": "此步骤的风险等级"
},
"reversible": {
"type": "boolean",
"description": "此步骤是否可以撤销"
}
},
"required": ["step_number", "action", "risk_level", "reversible"]
}
},
"missing_information": {
"type": "array",
"items": {"type": "string"},
"description": "执行任务缺少的关键信息列表（如有）"
},
"potential_issues": {
"type": "array",
"items": {"type": "string"},
"description": "预见到的潜在问题或风险"
},
"recommendation": {
"type": "string",
"enum": ["proceed", "clarify", "reject"],
"description": "执行建议：proceed（信息充分，可以执行）/ clarify（需要用户补充信息）/ reject（拒绝执行，理由见 rejection_reason）"
},
"rejection_reason": {
"type": "string",
"description": "如果 recommendation=reject，说明原因"
}
},
"required": ["task_description", "task_complexity", "proposed_steps", "recommendation"]
}
}
24.4 Advisor-Executor 架构
Advisor Tool 通常与执行工具组合使用，形成"先规划后执行"的双阶段架构：
import anthropic
import json
from typing import List, Dict, Any
class AdvisorExecutorAgent:
"""实现 Advisor-Executor 双阶段架构的 Agent"""
def __init__(
self,
execution_tools: List[dict],
tool_functions: Dict[str, callable],
model: str = "claude-opus-4-5",
require_approval_for: List[str] = None # 需要人工审批的风险等级
):
self.client = anthropic.Anthropic()
self.model = model
self.execution_tools = execution_tools
self.tool_functions = tool_functions
self.require_approval_for = require_approval_for or ["high", "critical"]
# Advisor 阶段：只有 advisor 工具
self.advisor_tools = [advisor_tool]
# Execution 阶段：advisor + 执行工具
self.all_tools = [advisor_tool] + execution_tools
self.execution_plan = None
self.advisor_result = None
def _run_advisor_phase(self, user_message: str, system: str) -> dict:
"""运行 Advisor 阶段，获取执行计划"""
print("\n" + "="*50)
print("ADVISOR 阶段：任务规划中...")
print("="*50)
messages = [{"role": "user", "content": user_message}]
response = self.client.messages.create(
model=self.model,
max_tokens=4096,
system=system + "\n\n在执行任何操作之前，必须先调用 advisor 工具进行任务规划。",
tools=self.advisor_tools, # 只有 advisor 工具
tool_choice={"type": "tool", "name": "advisor"}, # 强制使用 advisor
messages=messages
)
# 提取 advisor 调用结果
for block in response.content:
if block.type == "tool_use" and block.name == "advisor":
advisor_input = block.input
self.advisor_result = advisor_input
print(f"\n任务复杂度: {advisor_input.get('task_complexity', 'unknown')}")
print(f"执行建议: {advisor_input.get('recommendation', 'unknown')}")
steps = advisor_input.get('proposed_steps', [])
print(f"\n执行计划 ({len(steps)} 步):")
for step in steps:
risk_emoji = {"low": "✓", "medium": "△", "high": "⚠", "critical": "✗"}.get(
step.get("risk_level", "low"), "?"
)
print(f" 步骤{step['step_number']} [{risk_emoji}{step['risk_level']}] {step['action']}")
issues = advisor_input.get('potential_issues', [])
if issues:
print(f"\n潜在问题:")
for issue in issues:
print(f" - {issue}")
missing = advisor_input.get('missing_information', [])
if missing:
print(f"\n缺少信息:")
for info in missing:
print(f" - {info}")
return advisor_input
return {}
def _needs_approval(self, advisor_result: dict) -> bool:
"""检查是否需要人工审批"""
steps = advisor_result.get('proposed_steps', [])
for step in steps:
risk = step.get('risk_level', 'low')
if risk in self.require_approval_for:
return True
return False
def _request_approval(self, advisor_result: dict) -> bool:
"""请求人工审批"""
print("\n" + "⚠ "*20)
print("发现高风险步骤，需要人工审批")
print("⚠ "*20)
steps = advisor_result.get('proposed_steps', [])
risky_steps = [
s for s in steps
if s.get('risk_level') in self.require_approval_for
]
print(f"\n高风险步骤:")
for step in risky_steps:
print(f" - 步骤{step['step_number']}: {step['action']}")
print(f" 风险: {step['risk_level']}, 可逆: {step.get('reversible', '未知')}")
response = input("\n是否批准执行？(yes/no): ").strip().lower()
return response in ['yes', 'y']
def _run_execution_phase(self, user_message: str, system: str,
advisor_result: dict) -> str:
"""运行执行阶段"""
print("\n" + "="*50)
print("EXECUTION 阶段：任务执行中...")
print("="*50)
# 将执行计划注入系统提示
plan_summary = json.dumps(advisor_result, ensure_ascii=False, indent=2)
enhanced_system = f"""{system}
你已经通过 advisor 工具制定了以下执行计划：
{plan_summary}
请严格按照计划执行。如果实际执行中遇到计划之外的情况，先停下来重新评估。"""
messages = [{"role": "user", "content": user_message}]
while True:
response = self.client.messages.create(
model=self.model,
max_tokens=8192,
system=enhanced_system,
tools=self.all_tools,
messages=messages
)
if response.stop_reason == "end_turn":
return ' '.join(b.text for b in response.content if b.type == "text")
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type != "tool_use":
continue
if block.name == "advisor":
# 执行阶段中的二次规划（遇到意外情况）
print(f"\n[二次规划] 重新评估...")
result_content = json.dumps({
"acknowledged": True,
"message": "规划已记录，请按照新计划继续执行"
}, ensure_ascii=False)
else:
# 执行实际工具
print(f"\n[执行] {block.name}: {json.dumps(block.input, ensure_ascii=False)[:100]}")
try:
result = self.tool_functions[block.name](**block.input)
result_content = json.dumps(result, ensure_ascii=False, default=str)
print(f"[成功] {block.name}")
except Exception as e:
result_content = json.dumps({
"error": str(e),
"tool": block.name,
"input": block.input
}, ensure_ascii=False)
print(f"[失败] {block.name}: {e}")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result_content
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
else:
break
return "执行完成"
def run(self, user_message: str, system: str = "") -> str:
"""完整的 Advisor-Executor 流程"""
default_system = """你是一个谨慎且系统化的 AI 助手。
在执行任何复杂任务前，你会先系统地规划，评估风险，确保万无一失。"""
effective_system = system or default_system
# 阶段一：规划
advisor_result = self._run_advisor_phase(user_message, effective_system)
if not advisor_result:
return "规划阶段失败，无法继续执行"
# 检查建议
recommendation = advisor_result.get('recommendation', 'proceed')
if recommendation == 'reject':
reason = advisor_result.get('rejection_reason', '任务不适合自动执行')
return f"任务已拒绝执行。原因：{reason}"
if recommendation == 'clarify':
missing = advisor_result.get('missing_information', [])
questions = '\n'.join(f" - {q}" for q in missing)
return f"执行前需要以下信息：\n{questions}"
# 检查是否需要人工审批
if self._needs_approval(advisor_result):
if not self._request_approval(advisor_result):
return "任务已被用户取消"
# 阶段二：执行
return self._run_execution_phase(user_message, effective_system, advisor_result)
24.5 实战案例：自动化数据迁移任务
def demo_data_migration():
"""演示 Advisor 工具在高风险数据迁移任务中的应用"""
# 定义执行工具
execution_tools = [
{
"name": "list_database_tables",
"description": "列出数据库中的所有表",
"input_schema": {
"type": "object",
"properties": {
"database": {"type": "string"}
},
"required": ["database"]
}
},
{
"name": "backup_table",
"description": "备份指定数据表到备份存储",
"input_schema": {
"type": "object",
"properties": {
"table_name": {"type": "string"},
"backup_name": {"type": "string"}
},
"required": ["table_name", "backup_name"]
}
},
{
"name": "migrate_data",
"description": "将数据从源表迁移到目标表（高风险操作）",
"input_schema": {
"type": "object",
"properties": {
"source_table": {"type": "string"},
"target_table": {"type": "string"},
"transformation_sql": {"type": "string"}
},
"required": ["source_table", "target_table"]
}
},
{
"name": "verify_migration",
"description": "验证迁移数据的完整性",
"input_schema": {
"type": "object",
"properties": {
"source_table": {"type": "string"},
"target_table": {"type": "string"}
},
"required": ["source_table", "target_table"]
}
}
]
tool_functions = {
"list_database_tables": lambda database: {
"tables": ["users", "orders", "products", "analytics_old"]
},
"backup_table": lambda table_name, backup_name: {
"success": True,
"backup_path": f"/backups/{backup_name}.sql",
"row_count": 50000
},
"migrate_data": lambda source_table, target_table, transformation_sql=None: {
"success": True,
"migrated_rows": 50000,
"failed_rows": 0,
"duration_seconds": 12.5
},
"verify_migration": lambda source_table, target_table: {
"source_count": 50000,
"target_count": 50000,
"checksums_match": True
}
}
agent = AdvisorExecutorAgent(
execution_tools=execution_tools,
tool_functions=tool_functions,
require_approval_for=["high", "critical"]
)
result = agent.run(
user_message="""将生产数据库 'prod_db' 中的 'analytics_old' 表迁移到新的 'analytics_v2' 表。
迁移过程中需要将 timestamp 字段从 Unix 时间戳格式转换为 ISO 8601 格式。""",
system="""你是一个负责数据库运维的 AI 助手。
对于数据迁移任务，你需要格外谨慎，确保数据安全和完整性。"""
)
print("\n=== 最终结果 ===")
print(result)
return result
24.6 Advisor 与 Extended Thinking 的组合
Advisor Tool 本身适合与 Extended Thinking 配合，让规划过程更深入：
class ThinkingAdvisorAgent(AdvisorExecutorAgent):
"""Advisor 阶段使用 Extended Thinking 的增强版 Agent"""
def _run_advisor_phase(self, user_message: str, system: str) -> dict:
"""使用 Extended Thinking 的 Advisor 阶段"""
messages = [{"role": "user", "content": user_message}]
response = self.client.messages.create(
model=self.model,
max_tokens=16000,
thinking={
"type": "enabled",
"budget_tokens": 8000
},
system=system + "\n\n首先深度思考任务的各个方面，然后调用 advisor 工具输出结构化规划。",
tools=self.advisor_tools,
tool_choice={"type": "tool", "name": "advisor"},
messages=messages
)
# 显示思维过程摘要
for block in response.content:
if block.type == "thinking":
print(f"\n[深度思考] {len(block.thinking)} 字符")
print(f"思维预览: {block.thinking[:200]}...")
elif block.type == "tool_use" and block.name == "advisor":
self.advisor_result = block.input
return block.input
return {}
24.7 元认知反馈循环：任务完成后的自评
Advisor 模式可以扩展为完整的元认知反馈循环：
post_execution_review_tool = {
"name": "post_execution_review",
"description": """任务执行完成后调用此工具进行自我评估。
评估内容包括：实际执行是否符合计划、遇到了哪些预期外情况、下次如何改进。""",
"input_schema": {
"type": "object",
"properties": {
"planned_steps": {
"type": "integer",
"description": "计划的步骤数量"
},
"actual_steps": {
"type": "integer",
"description": "实际执行的步骤数量"
},
"deviations": {
"type": "array",
"items": {"type": "string"},
"description": "实际执行与计划的偏差列表"
},
"unexpected_issues": {
"type": "array",
"items": {"type": "string"},
"description": "执行中遇到的意外问题"
},
"task_success": {
"type": "boolean",
"description": "任务是否成功完成"
},
"lessons_learned": {
"type": "array",
"items": {"type": "string"},
"description": "从此次执行中学到的经验"
},
"quality_score": {
"type": "integer",
"minimum": 1,
"maximum": 10,
"description": "对本次执行质量的自评分（1-10）"
}
},
"required": ["task_success", "quality_score"]
}
}
24.8 Advisor 模式的适用场景决策树
任务是否明确且简单？
├── 是 → 直接执行，无需 Advisor
└── 否 ↓
任务步骤数 > 3？
├── 否 → 简单 Advisor（只需检查依赖关系）
└── 是 ↓
任务是否涉及不可逆操作？
├── 否 → 标准 Advisor（规划 + 风险评估）
└── 是 ↓
任务失败的代价如何？
├── 低（可以重做）→ Advisor + 自动执行
├── 中（有代价）→ Advisor + 关键步骤确认
└── 高（灾难性）→ Advisor + 完整人工审批
def select_advisor_mode(task: str) -> str:
"""根据任务特征选择 Advisor 模式"""
irreversible_keywords = ["删除", "发送", "提交", "付款", "迁移", "格式化",
"delete", "send", "submit", "pay", "migrate", "format"]
complex_keywords = ["步骤", "然后", "接着", "之后", "最后",
"step", "then", "next", "after", "finally"]
task_lower = task.lower()
is_irreversible = any(kw in task_lower for kw in irreversible_keywords)
is_complex = any(kw in task_lower for kw in complex_keywords) or len(task) > 200
if not is_complex and not is_irreversible:
return "direct" # 直接执行
elif is_complex and not is_irreversible:
return "light_advisor" # 轻量规划
elif not is_complex and is_irreversible:
return "safety_advisor" # 安全确认
else:
return "full_advisor" # 完整规划 + 审批
小结
Advisor Tool 模式将 Claude 的元认知能力转化为可观测、可审计的工具调用，为复杂任务提供了"先规划、后执行"的安全保障。核心价值在于：
- 透明性：执行计划以结构化数据呈现，人类可以理解和审查
- 可控性：
recommendation字段和风险等级使得人工干预点清晰可见 - 安全性：不可逆操作必须经过显式规划和可选的人工审批
- 可改进性：通过 post_execution_review，建立元认知反馈循环
结合 Extended Thinking，Advisor Tool 让 Claude 在最高风险的任务上展现出最深度的思考和最系统的规划能力。
至此，《Claude 完全指南》的 Part 4 Tool Use 与函数调用章节已全部完成。下一部分将探讨多智能体架构、提示词工程的高级技巧，以及 Claude 在生产环境中的安全合规实践。