第 22 章
Programmatic Tool Calling：代码执行容器内直接调用工具减少 Round-Trip
第二十二章：Tool Use + Extended Thinking：推理与行动的最强组合
22.1 为何要组合这两种能力
Tool Use 和 Extended Thinking 各自解决了不同的问题：
- Tool Use 赋予 Claude 与外部世界交互的能力，突破了训练数据的局限
- Extended Thinking 赋予 Claude 深度推理能力，使其在复杂问题上避免浅层思考
但真正的复杂任务往往同时需要两者：先深度思考应该调用哪些工具、如何组合工具的结果，然后执行工具调用，再对结果进行深度分析。
这种组合在以下场景中尤为强大：
- 复杂数据分析：先思考分析框架，再查询数据，再深度解读结果
- 多步骤问题解决：推理解题路径，调用工具执行，验证结果
- 策略规划：思考约束条件，调用工具收集信息，制定最优方案
- 代码生成与验证：思考架构设计，生成代码，执行测试，分析失败原因
22.2 API 配置：同时启用两种能力
同时使用 Tool Use 和 Extended Thinking 的配置方式：
import anthropic
import json
client = anthropic.Anthropic()
# 定义工具
analysis_tools = [
{
"name": "query_database",
"description": "查询数据库获取统计数据",
"input_schema": {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "SQL 查询语句"
},
"database": {
"type": "string",
"enum": ["sales", "users", "products"],
"description": "目标数据库"
}
},
"required": ["sql", "database"]
}
},
{
"name": "run_python_code",
"description": "执行 Python 代码进行数据分析和可视化",
"input_schema": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "要执行的 Python 代码"
}
},
"required": ["code"]
}
}
]
# 同时启用 Extended Thinking 和 Tool Use
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=16000,
thinking={
"type": "enabled",
"budget_tokens": 8000 # 为每次推理预留 8000 token
},
tools=analysis_tools,
messages=[{
"role": "user",
"content": """分析我们过去三个月的销售数据，找出增长最快的产品类别，
并给出下季度的库存优化建议。请深入思考后给出有据可查的结论。"""
}]
)
# 响应中可能包含 thinking block、tool_use block 和 text block
for block in response.content:
print(f"Block type: {block.type}")
22.3 thinking block 在工具调用周期中的分布
在 Tool Use + Extended Thinking 组合中，thinking block 可能出现在多个位置：
第一轮（初始推理）:
thinking: "让我先思考这个问题的分析框架..."
tool_use: query_database(...)
第二轮（结果分析推理）:
tool_result: {...查询结果...}
thinking: "看到这些数据，我需要进一步分析..."
tool_use: run_python_code(...)
第三轮（最终推理与综合）:
tool_result: {...计算结果...}
thinking: "综合所有数据，我可以得出以下结论..."
text: "根据深入分析，以下是我的建议..."
def analyze_thinking_distribution(all_messages: list) -> dict:
"""分析 thinking block 在工具调用周期中的分布"""
distribution = {
"pre_tool_thinking": [], # 工具调用前的思维
"post_result_thinking": [], # 获取结果后的思维
"final_thinking": [] # 生成最终答案前的思维
}
for msg in all_messages:
if msg["role"] != "assistant":
continue
content = msg["content"]
if not isinstance(content, list):
continue
has_tool_use = any(
(b.type == "tool_use" if hasattr(b, 'type') else b.get("type") == "tool_use")
for b in content
)
has_text = any(
(b.type == "text" if hasattr(b, 'type') else b.get("type") == "text")
for b in content
)
for block in content:
block_type = block.type if hasattr(block, 'type') else block.get("type")
if block_type == "thinking":
thinking_text = block.thinking if hasattr(block, 'thinking') else block.get("thinking", "")
if has_tool_use and not has_text:
distribution["pre_tool_thinking"].append(len(thinking_text))
elif has_text and not has_tool_use:
distribution["final_thinking"].append(len(thinking_text))
else:
distribution["post_result_thinking"].append(len(thinking_text))
return {
k: {
"count": len(v),
"avg_chars": sum(v) / len(v) if v else 0,
"total_chars": sum(v)
}
for k, v in distribution.items()
}
22.4 完整实现：数据分析智能体
import anthropic
import json
from typing import Any, List
class ThinkingToolAgent:
"""同时使用 Extended Thinking 和 Tool Use 的智能体"""
def __init__(
self,
model: str = "claude-opus-4-5",
thinking_budget: int = 8000,
max_tokens: int = 20000
):
self.client = anthropic.Anthropic()
self.model = model
self.thinking_budget = thinking_budget
self.max_tokens = max_tokens
self.tools = []
self.tool_functions = {}
self.conversation_history = []
self.thinking_log = []
def add_tool(self, tool_definition: dict, func):
"""注册工具"""
self.tools.append(tool_definition)
self.tool_functions[tool_definition["name"]] = func
return self
def _execute_tool(self, name: str, inputs: dict) -> Any:
"""执行工具并处理异常"""
if name not in self.tool_functions:
raise ValueError(f"工具未注册: {name}")
return self.tool_functions[name](**inputs)
def _build_tool_result(self, tool_use_id: str, result: Any,
error: bool = False) -> dict:
"""构建工具结果块"""
content = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False, default=str)
block = {
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": content
}
if error:
block["is_error"] = True
return block
def _log_thinking(self, thinking_text: str, round_num: int):
"""记录思维过程"""
self.thinking_log.append({
"round": round_num,
"chars": len(thinking_text),
"preview": thinking_text[:300] + "..." if len(thinking_text) > 300 else thinking_text
})
def run(self, user_message: str, system: str = "") -> str:
"""执行完整的推理+工具调用循环"""
self.conversation_history = [{"role": "user", "content": user_message}]
create_kwargs = {
"model": self.model,
"max_tokens": self.max_tokens,
"thinking": {
"type": "enabled",
"budget_tokens": self.thinking_budget
},
"tools": self.tools,
"messages": self.conversation_history
}
if system:
create_kwargs["system"] = system
round_num = 0
while round_num < 10:
round_num += 1
print(f"\n=== 第 {round_num} 轮 ===")
response = self.client.messages.create(**create_kwargs)
# 处理 thinking blocks
thinking_chars = 0
tool_calls = []
for block in response.content:
if block.type == "thinking":
thinking_chars += len(block.thinking)
self._log_thinking(block.thinking, round_num)
print(f"[思考] {len(block.thinking)} 字符")
elif block.type == "tool_use":
tool_calls.append(block)
print(f"[工具] {block.name}: {json.dumps(block.input, ensure_ascii=False)[:100]}")
elif block.type == "text":
print(f"[文本] {block.text[:100]}...")
# 完成
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 tool_call in tool_calls:
try:
result = self._execute_tool(tool_call.name, tool_call.input)
tool_results.append(
self._build_tool_result(tool_call.id, result)
)
print(f"[结果] {tool_call.name}: 成功")
except Exception as e:
tool_results.append(
self._build_tool_result(tool_call.id, str(e), error=True)
)
print(f"[结果] {tool_call.name}: 失败 - {e}")
# 更新消息历史（保留完整 content，包括 thinking blocks）
self.conversation_history.append({
"role": "assistant",
"content": response.content
})
self.conversation_history.append({
"role": "user",
"content": tool_results
})
create_kwargs["messages"] = self.conversation_history
return "达到最大迭代次数"
def get_thinking_summary(self) -> str:
"""获取思维过程摘要"""
if not self.thinking_log:
return "没有思维记录"
total_chars = sum(t["chars"] for t in self.thinking_log)
summary = f"共 {len(self.thinking_log)} 次思维，总计 {total_chars} 字符\n"
for entry in self.thinking_log:
summary += f"\n第{entry['round']}轮（{entry['chars']}字符）:\n"
summary += f" {entry['preview']}\n"
return summary
22.5 实战案例：复杂投资分析
def build_investment_analysis_agent():
"""构建复杂投资分析智能体"""
agent = ThinkingToolAgent(
thinking_budget=12000,
max_tokens=24000
)
# 注册数据获取工具
def get_stock_data(symbol: str, period: str = "1y") -> dict:
"""获取股票历史数据（模拟）"""
return {
"symbol": symbol,
"period": period,
"data": [
{"date": "2026-01-01", "close": 150.0, "volume": 1000000},
{"date": "2026-02-01", "close": 162.0, "volume": 1200000},
{"date": "2026-03-01", "close": 158.0, "volume": 900000},
{"date": "2026-04-01", "close": 175.0, "volume": 1500000},
],
"current_price": 175.0,
"52w_high": 185.0,
"52w_low": 130.0
}
def get_financial_statements(symbol: str, statement_type: str) -> dict:
"""获取财务报表（模拟）"""
return {
"symbol": symbol,
"type": statement_type,
"revenue": 5000000000,
"net_income": 800000000,
"eps": 4.5,
"pe_ratio": 38.9,
"debt_to_equity": 0.45
}
def get_analyst_ratings(symbol: str) -> dict:
"""获取分析师评级（模拟）"""
return {
"symbol": symbol,
"consensus": "Buy",
"target_price": 200.0,
"num_analysts": 28,
"buy": 18,
"hold": 8,
"sell": 2
}
def calculate_valuation_metrics(
current_price: float,
eps: float,
growth_rate: float,
discount_rate: float
) -> dict:
"""计算估值指标"""
pe = current_price / eps if eps > 0 else 0
# 简化的 DCF 模型
dcf_value = sum(
eps * (1 + growth_rate) ** i / (1 + discount_rate) ** i
for i in range(1, 6)
)
return {
"current_pe": round(pe, 2),
"dcf_value": round(dcf_value, 2),
"upside_potential": round((dcf_value - current_price) / current_price * 100, 1)
}
agent.add_tool({
"name": "get_stock_data",
"description": "获取股票历史价格和基本行情数据",
"input_schema": {
"type": "object",
"properties": {
"symbol": {"type": "string"},
"period": {"type": "string", "enum": ["1m", "3m", "6m", "1y", "3y"]}
},
"required": ["symbol"]
}
}, get_stock_data)
agent.add_tool({
"name": "get_financial_statements",
"description": "获取公司财务报表数据（利润表、资产负债表等）",
"input_schema": {
"type": "object",
"properties": {
"symbol": {"type": "string"},
"statement_type": {"type": "string", "enum": ["income", "balance_sheet", "cash_flow"]}
},
"required": ["symbol", "statement_type"]
}
}, get_financial_statements)
agent.add_tool({
"name": "get_analyst_ratings",
"description": "获取华尔街分析师的评级和目标价",
"input_schema": {
"type": "object",
"properties": {
"symbol": {"type": "string"}
},
"required": ["symbol"]
}
}, get_analyst_ratings)
agent.add_tool({
"name": "calculate_valuation_metrics",
"description": "基于输入参数计算估值指标，包括 P/E 和简化 DCF 模型",
"input_schema": {
"type": "object",
"properties": {
"current_price": {"type": "number"},
"eps": {"type": "number"},
"growth_rate": {"type": "number", "description": "预期增长率（小数）"},
"discount_rate": {"type": "number", "description": "折现率（小数）"}
},
"required": ["current_price", "eps", "growth_rate", "discount_rate"]
}
}, calculate_valuation_metrics)
return agent
# 使用示例
agent = build_investment_analysis_agent()
result = agent.run(
user_message="""请对 NVDA（英伟达）进行深度投资分析，包括：
1. 近期价格走势和技术面分析
2. 基本面数据（营收、利润、估值指标）
3. 分析师观点
4. 基于 DCF 模型的内在价值估算（假设未来5年增长率20%，折现率10%）
5. 综合投资建议（买入/持有/卖出）及理由
请深入思考每个步骤的逻辑，确保结论有数据支撑。""",
system="""你是一个专业的股票分析师，擅长从多个维度深入分析股票投资价值。
在分析前请先思考分析框架，确保覆盖全面且逻辑严密。"""
)
print("\n=== 投资分析报告 ===")
print(result)
print("\n=== 思维过程摘要 ===")
print(agent.get_thinking_summary())
22.6 优化：thinking budget 的动态分配
不同阶段的推理需要不同深度的思考：
class AdaptiveThinkingAgent(ThinkingToolAgent):
"""自适应 thinking budget 的智能体"""
THINKING_PROFILES = {
"quick": 2000, # 快速判断
"standard": 5000, # 标准推理
"deep": 10000, # 深度分析
"maximum": 20000 # 最大深度
}
def __init__(self, default_profile: str = "standard"):
super().__init__(
thinking_budget=self.THINKING_PROFILES[default_profile]
)
self.default_profile = default_profile
def run_with_adaptive_budget(self, user_message: str,
complexity_hint: str = "auto") -> str:
"""根据任务复杂度自动调整 thinking budget"""
if complexity_hint == "auto":
# 基于问题特征判断复杂度
question_len = len(user_message)
has_numbers = any(c.isdigit() for c in user_message)
has_comparison = any(w in user_message for w in ["比较", "对比", "分析", "评估"])
has_multi_step = any(w in user_message for w in ["然后", "接着", "步骤", "流程"])
complexity_score = 0
complexity_score += min(question_len // 100, 3)
complexity_score += 2 if has_numbers else 0
complexity_score += 2 if has_comparison else 0
complexity_score += 2 if has_multi_step else 0
if complexity_score <= 2:
profile = "quick"
elif complexity_score <= 4:
profile = "standard"
elif complexity_score <= 6:
profile = "deep"
else:
profile = "maximum"
else:
profile = complexity_hint
self.thinking_budget = self.THINKING_PROFILES.get(profile, 5000)
print(f"使用 thinking profile: {profile} ({self.thinking_budget} tokens)")
return self.run(user_message)
22.7 特殊模式：先规划再执行
对于复杂的多步骤任务，可以先让 Claude 在 thinking 中规划完整路径：
def run_plan_then_execute(agent: ThinkingToolAgent, task: str) -> str:
"""先让 Claude 制定完整计划，再逐步执行"""
# 第一轮：纯规划（不调用工具）
planning_response = agent.client.messages.create(
model=agent.model,
max_tokens=agent.max_tokens,
thinking={
"type": "enabled",
"budget_tokens": agent.thinking_budget
},
# 不传工具，强制 Claude 先思考规划
messages=[{
"role": "user",
"content": f"""请先制定一个完整的执行计划（不要执行，只规划）：
任务：{task}
请用以下格式输出计划：
## 分析框架
（你打算如何分析这个问题）
## 执行步骤
1. （第一步）
2. （第二步）
...
## 预期结论结构
（最终报告应该包含哪些部分）"""
}]
)
# 提取计划
plan_text = ' '.join(
b.text for b in planning_response.content if b.type == "text"
)
print("=== 执行计划 ===")
print(plan_text)
# 第二轮：根据计划执行（携带工具）
result = agent.run(
user_message=f"""根据以下计划执行任务：
{plan_text}
原始任务：{task}""",
system="严格按照制定的计划执行，使用工具获取所需数据，最后给出完整报告。"
)
return result
小结
Tool Use + Extended Thinking 的组合代表了目前 Claude API 最高级的使用模式。核心原则：
- thinking 在工具调用前：帮助 Claude 规划工具使用策略
- thinking 在结果获取后：帮助 Claude 深度解读和整合数据
- thinking budget 要充足：推理+工具调用的场景通常需要 5000-15000 token
- 保留完整历史：包含 thinking blocks 的消息历史必须完整传递，维护推理连续性
在后续章节中，我们将探索更高级的工具使用模式：动态工具发现（Tool Search）和 Claude 的自我规划能力（Advisor Tool）。