企业官网建设流程全解析

在实际 AI 应用开发中，Agent 的核心能力往往被包装在复杂的框架和 SDK 中，导致很多开发者只知其名、不明其理。真正理解 Agent 如何调用工具、处理结果、维持会话，比单纯学会使用某个框架更有价值。本文将围绕 Claude Platform 文档中提到的tool_use、tool_result和messages机制，用约 30 行 Python 代码实现一个最小可运行的 Agent 核心逻辑，帮助读者掌握工具调用的完整生命周期。

这个简化版 Agent 将展示：如何接收用户请求、解析出需要调用的工具、执行具体函数、返回工具结果，并继续后续对话。虽然生产级 Agent 需要考虑上下文管理、错误处理、异步调用等复杂问题，但这个最小实现能让你看清最本质的交互流程。

1. 先理解 Agent 工具调用的三个关键环节

Agent 的核心是能够根据当前对话上下文，决定是否需要调用外部工具（如计算器、搜索引擎、数据库查询等），并将工具执行结果整合回对话流。这个流程涉及三个关键环节：

1.1 tool_use：Agent 决定调用哪个工具

当 Agent 判断需要工具协助时，会在回复中插入tool_use块，明确指定要调用的工具名称和传入参数。这相当于 Agent 的"决策输出"。

{ "type": "tool_use", "id": "toolu_01", "name": "calculate", "input": {"expression": "2 + 3 * 4"} }

1.2 tool_result：执行工具并返回结果

应用程序收到tool_use后，需要找到对应的工具函数执行，然后将执行结果包装成tool_result格式返回给 Agent。这相当于"工具执行反馈"。

{ "type": "tool_result", "tool_use_id": "toolu_01", "content": "14" }

1.3 messages：维持完整的对话上下文

所有的用户输入、Agent 回复、工具调用和工具结果都以消息形式保存在会话中。Agent 基于完整的消息历史来决定下一步行动，这是维持连贯对话的关键。

2. 准备最小化开发环境

这个实现只需要 Python 3.7+ 和 requests 库，不需要安装任何 AI 框架或 SDK。

2.1 环境要求

2.2 安装依赖

pip install requests

2.3 设置 API 密钥

将你的 Claude API 密钥设置为环境变量：

export ANTHROPIC_API_KEY='your-api-key-here'

或者在代码中直接设置：

import os os.environ['ANTHROPIC_API_KEY'] = 'your-api-key-here'

3. 实现 30 行核心 Agent 逻辑

下面是最小化 Agent 核心实现，完整代码约 30 行：

import json import requests from typing import Dict, Any, List class MiniAgent: def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.anthropic.com/v1/messages" self.messages = [] def add_message(self, role: str, content: str): self.messages.append({"role": role, "content": content}) def call_tool(self, tool_name: str, input_data: Dict) -> str: tools = { "calculate": lambda x: str(eval(x["expression"])), "get_weather": lambda x: f"Sunny, {x.get('temperature', '25°C')}", "search_web": lambda x: f"Results for: {x.get('query', '')}" } return tools.get(tool_name, lambda x: "Tool not found")(input_data) def send_request(self, tools: List[Dict]) -> Dict: headers = { "x-api-key": self.api_key, "anthropic-version": "2023-06-01", "content-type": "application/json" } data = { "model": "claude-3-haiku-20240307", "max_tokens": 1000, "messages": self.messages, "tools": tools } response = requests.post(self.base_url, headers=headers, json=data) return response.json() def process(self, user_input: str, tools: List[Dict]) -> str: self.add_message("user", user_input) response = self.send_request(tools) if "content" in response: for block in response["content"]: if block["type"] == "text": return block["text"] elif block["type"] == "tool_use": tool_result = self.call_tool(block["name"], block["input"]) self.add_message("assistant", json.dumps([block])) self.add_message("user", json.dumps([{ "type": "tool_result", "tool_use_id": block["id"], "content": tool_result }])) return self.process("Continue", tools) return "No response"

4. 关键代码详解：理解每个组件的作用

4.1 消息管理：维持对话上下文

messages列表保存完整的对话历史，每条消息包含角色（user/assistant）和内容。这是 Agent 理解上下文的基础：

def add_message(self, role: str, content: str): self.messages.append({"role": role, "content": content})

在工具调用场景中，需要特殊处理消息格式：

• Assistant 的消息可能包含tool_use块
• User 的消息可能包含tool_result块

4.2 工具注册与执行：扩展 Agent 能力

call_tool方法维护一个工具字典，将工具名映射到具体的执行函数：

def call_tool(self, tool_name: str, input_data: Dict) -> str: tools = { "calculate": lambda x: str(eval(x["expression"])), # 简单计算器 "get_weather": lambda x: f"Sunny, {x.get('temperature', '25°C')}", "search_web": lambda x: f"Results for: {x.get('query', '')}" } return tools.get(tool_name, lambda x: "Tool not found")(input_data)

实际项目中，这些工具函数可以替换为真实的 API 调用、数据库查询或复杂计算逻辑。

4.3 API 请求封装：与 Claude 模型交互

send_request方法封装了与 Claude API 的通信细节：

def send_request(self, tools: List[Dict]) -> Dict: headers = { "x-api-key": self.api_key, "anthropic-version": "2023-06-01", # 指定 API 版本 "content-type": "application/json" } data = { "model": "claude-3-haiku-20240307", # 使用轻量级模型 "max_tokens": 1000, # 限制响应长度 "messages": self.messages, # 传入完整对话历史 "tools": tools # 定义可用的工具列表 } response = requests.post(self.base_url, headers=headers, json=data) return response.json()

关键参数说明：

• model: 选择适合的 Claude 模型版本
• max_tokens: 控制响应长度，避免过长回复
• tools: 定义 Agent 可以使用的工具列表

4.4 核心处理逻辑：工具调用循环

process方法实现了完整的工具调用生命周期：

def process(self, user_input: str, tools: List[Dict]) -> str: # 1. 添加用户消息到历史 self.add_message("user", user_input) # 2. 发送请求到 Claude API response = self.send_request(tools) # 3. 处理响应内容 if "content" in response: for block in response["content"]: if block["type"] == "text": # 如果是普通文本回复，直接返回 return block["text"] elif block["type"] == "tool_use": # 如果是工具调用请求 # 4. 执行工具函数 tool_result = self.call_tool(block["name"], block["input"]) # 5. 将工具调用和结果添加到消息历史 self.add_message("assistant", json.dumps([block])) self.add_message("user", json.dumps([{ "type": "tool_result", "tool_use_id": block["id"], "content": tool_result }])) # 6. 递归调用，继续处理工具结果 return self.process("Continue", tools) return "No response"

这个循环确保了：Agent 决定调用工具 → 执行工具 → 返回结果 → Agent 基于结果继续思考的完整流程。

5. 运行完整示例：从数学计算到天气查询

5.1 定义可用工具列表

首先定义 Agent 可以使用的工具：

tools = [ { "name": "calculate", "description": "Evaluate a mathematical expression", "input_schema": { "type": "object", "properties": { "expression": {"type": "string", "description": "Math expression to evaluate"} }, "required": ["expression"] } }, { "name": "get_weather", "description": "Get current weather information", "input_schema": { "type": "object", "properties": { "location": {"type": "string", "description": "City name"}, "temperature": {"type": "string", "description": "Temperature unit"} }, "required": ["location"] } } ]

5.2 初始化并运行 Agent

# 初始化 Agent（API 密钥需要替换为实际值） agent = MiniAgent(api_key=os.getenv("ANTHROPIC_API_KEY")) # 测试数学计算 result1 = agent.process("Calculate 15 * 8 + 20", tools) print(f"Result 1: {result1}") # 测试天气查询（会调用工具） result2 = agent.process("What's the weather in Beijing?", tools) print(f"Result 2: {result2}") # 查看完整的消息历史 print("Message history:") for msg in agent.messages: print(f"{msg['role']}: {msg['content'][:100]}...")

5.3 预期输出示例

Result 1: 15 * 8 + 20 = 140 Result 2: The weather in Beijing is sunny with a temperature of 25°C. Message history: user: Calculate 15 * 8 + 20 assistant: [{"type": "tool_use", "id": "toolu_01", "name": "calculate", "input": {"expression": "15 * 8 + 20"}}] user: [{"type": "tool_result", "tool_use_id": "toolu_01", "content": "140"}] assistant: The result of 15 * 8 + 20 is 140. user: What's the weather in Beijing? assistant: [{"type": "tool_use", "id": "toolu_02", "name": "get_weather", "input": {"location": "Beijing"}}] user: [{"type": "tool_result", "tool_use_id": "toolu_02", "content": "Sunny, 25°C"}] assistant: The weather in Beijing is sunny with a temperature of 25°C.

6. 常见问题排查与解决方案

在实际运行过程中，可能会遇到以下典型问题：

6.1 API 认证失败

现象：收到 401 错误或认证失败提示。

排查步骤：

1. 检查ANTHROPIC_API_KEY环境变量是否设置正确
2. 确认 API 密钥是否有调用 Messages API 的权限
3. 验证 API 密钥是否过期或被撤销

解决方案：

# 验证 API 密钥格式（以 sk-ant- 开头） api_key = os.getenv("ANTHROPIC_API_KEY") if not api_key or not api_key.startswith("sk-ant-"): print("Invalid API key format")

6.2 工具调用格式错误

现象：Agent 返回了tool_use块，但执行时报参数错误。

排查步骤：

1. 检查工具定义中的input_schema与实际调用参数是否匹配
2. 验证工具函数是否能处理传入的参数类型
3. 查看 Claude 响应中的tool_use块内容

解决方案：

# 添加参数验证 def call_tool(self, tool_name: str, input_data: Dict) -> str: try: tools = { "calculate": lambda x: str(eval(x.get("expression", "0"))), # 其他工具... } return tools[tool_name](input_data) except KeyError: return f"Tool {tool_name} not found" except Exception as e: return f"Tool error: {str(e)}"

6.3 消息历史过长导致 API 限制

现象：随着对话进行，响应变慢或出现 token 超限错误。

排查步骤：

1. 监控messages列表的长度和内容大小
2. 检查每次 API 调用的 token 使用量
3. 确认模型上下文窗口限制

解决方案：

# 实现消息历史截断策略 def truncate_messages(self, max_messages=20): if len(self.messages) > max_messages: # 保留最新的消息，但确保工具调用相关的消息完整 self.messages = self.messages[-max_messages:]

6.4 工具执行结果格式错误

现象：工具返回结果后，Agent 无法正确理解或处理。

排查步骤：

1. 检查tool_result的格式是否符合 Claude API 要求
2. 验证返回内容是否为字符串类型
3. 确认tool_use_id与对应的tool_use块匹配

解决方案：

# 确保工具结果格式正确 tool_result_block = { "type": "tool_result", "tool_use_id": block["id"], # 必须与 tool_use 的 id 匹配 "content": str(tool_result) # 确保内容为字符串 }

7. 生产环境最佳实践

虽然这个 30 行实现展示了核心逻辑，但生产环境需要考虑更多因素：

7.1 工具管理规范化

不要使用简单的字典映射，而是建立工具注册机制：

class ToolRegistry: def __init__(self): self._tools = {} def register(self, name: str, description: str, func: callable, schema: Dict): self._tools[name] = { "function": func, "description": description, "schema": schema } def get_tool_definitions(self): return [{ "name": name, "description": info["description"], "input_schema": info["schema"] } for name, info in self._tools.items()]

7.2 错误处理与重试机制

添加完善的异常处理：

def safe_call_tool(self, tool_name: str, input_data: Dict) -> str: try: result = self.call_tool(tool_name, input_data) return result except Exception as e: logger.error(f"Tool {tool_name} failed: {e}") return f"Error executing tool: {str(e)}" def send_request_with_retry(self, tools: List[Dict], max_retries=3): for attempt in range(max_retries): try: return self.send_request(tools) except requests.exceptions.RequestException as e: if attempt == max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避

7.3 会话状态管理

实现会话持久化和恢复：

class SessionManager: def __init__(self, storage_path: str): self.storage_path = storage_path def save_session(self, session_id: str, messages: List[Dict]): with open(f"{self.storage_path}/{session_id}.json", "w") as f: json.dump({"messages": messages}, f) def load_session(self, session_id: str) -> List[Dict]: try: with open(f"{self.storage_path}/{session_id}.json", "r") as f: return json.load(f)["messages"] except FileNotFoundError: return []

7.4 性能监控与限流

添加基本的监控指标：

class MonitoringAgent(MiniAgent): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.metrics = { "api_calls": 0, "tool_calls": 0, "total_tokens": 0 } def send_request(self, tools: List[Dict]) -> Dict: self.metrics["api_calls"] += 1 response = super().send_request(tools) if "usage" in response: self.metrics["total_tokens"] += response["usage"]["total_tokens"] return response

8. 扩展方向与深入学习建议

掌握这个核心逻辑后，可以从以下几个方向深入：

8.1 工具生态扩展

• 数据库工具：添加 SQL 查询、数据统计等工具
• API 集成：连接外部服务如日历、邮件、支付等
• 文件操作：实现文档处理、图片分析等能力
• 自定义计算：添加领域特定的计算和推理工具

8.2 高级 Agent 模式

• 多步推理：让 Agent 能够规划复杂的多步任务
• 工具组合：支持多个工具的顺序或并行执行
• 动态工具加载：根据上下文动态注册和卸载工具
• 工具学习：基于使用反馈优化工具选择策略

8.3 工程化改进

• 异步处理：使用 async/await 提高并发性能
• 缓存机制：对常用工具结果进行缓存
• 流量控制：实现 API 调用速率限制和配额管理
• 可观测性：添加完整的日志、指标和追踪

这个 30 行实现虽然简单，但包含了 Agent 工具调用的核心范式。理解这个基础后，再学习 LangChain、AutoGPT 等框架时会更容易把握其设计理念和实现原理。实际项目中，建议从这种最小可行实现开始，逐步添加所需的功能和优化。