引言

大语言模型的 API 正在改变我们构建应用的方式。Anthropic 的 Claude 系列以强大的推理能力、超长上下文窗口和优秀的指令遵循能力，成为许多开发者构建智能助手时的首选。不过，直接调用 API 只是第一步——如何让助手拥有记忆、支持流式输出、安全调用外部工具，才是真正的生产力落地。

本文将带你从零开始，使用 Claude API（Messages API）构建一个功能完整的智能助手。我们会覆盖：
- 如何设计消息格式与系统提示
- 如何实现多轮对话与上下文管理
- 流式输出与错误重试
- 让模型决定何时调用外部工具（函数调用）
- 敏感内容过滤与安全最佳实践

所有代码均基于 Python，完整可运行，文件结构清晰，注释详尽。读完本文，你就能将这套助手直接集成到自己的产品中。

核心概念：Messages API 的工作方式

Claude API 使用的是 Messages API，它围绕“消息列表”这一核心概念设计。每次请求你都要提供一个 messages 数组，其中每个元素包含 role（user 或 assistant）和 content。此外还可以设置 system 参数，用于全局级别的系统提示。

关键参数：
- model：模型版本，如 claude-3-5-sonnet-20240620
- max_tokens：模型最大输出 token 数
- temperature：控制随机性 (0-1)
- system：系统级指令，定义助手的行为、角色、知识边界等
- messages：对话历史，按时间顺序排列，必须以 user 角色结尾
- tools：可选，定义函数调用工具列表
- stream：布尔值，是否启用流式响应

理解这些基础后，我们直接开始构建。

实战：构建一个具备记忆与工具调用能力的助手

我们将实现一个命令行对话助手，它能够：
1. 记住完整的对话上下文
2. 在用户要求时查询天气（模拟工具调用）
3. 使用流式输出，逐字显示回复
4. 遇到超时或限流自动重试

环境准备

首先安装依赖：

pip install anthropic python-dotenv

在 .env 文件中设置你的 API 密钥：

ANTHROPIC_API_KEY=your-api-key-here

完整代码实现

新建 claude_assistant.py：

```python
import os
import json
import time
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
import anthropic

load_dotenv()

class ClaudeAssistant:
"""
一个具备上下文记忆与工具调用的 Claude 智能助手
"""
def init(
self,
system_prompt: str = "You are a helpful assistant.",
model: str = "claude-3-5-sonnet-20240620",
max_tokens: int = 1024,
temperature: float = 0.7,
max_history_tokens: int = 8000 # 防止上下文超长
):
self.client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
self.system_prompt = system_prompt
self.model = model
self.max_tokens = max_tokens
self.temperature = temperature
self.max_history_tokens = max_history_tokens
self.conversation: List[Dict[str, Any]] = [] # 对话历史

    # 定义助手可用的工具：查询天气（示例）
    self.tools = [
        {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "input_schema": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如 Beijing"
                    }
                },
                "required": ["city"]
            }
        }
    ]

def _manage_history(self):
    """
    简单的上下文管理：当历史消息估算 token 数超过阈值时，
    移除最早的几轮对话以保持上下文在可控范围内。
    （这里使用粗略估算：1 token ≈ 4 字符）
    """
    total_chars = sum(len(msg.get("content", "")) for msg in self.conversation)
    estimated_tokens = total_chars // 4
    while estimated_tokens > self.max_history_tokens and len(self.conversation) > 1:
        # 至少保留最后一轮对话（即最新的 user 和 assistant 消息）
        if len(self.conversation) >= 2 and self.conversation[0]["role"] == "user":
            self.conversation.pop(0)
            if self.conversation and self.conversation[0]["role"] == "assistant":
                self.conversation.pop(0)
        else:
            # 异常情况，直接弹出第一条
            self.conversation.pop(0)
        total_chars = sum(len(msg.get("content", "")) for msg in self.conversation)
        estimated_tokens = total_chars // 4

def _call_api(
    self,
    messages: List[Dict[str, Any]],
    tools: Optional[List[Dict]] = None,
    stream: bool = True
) -> anthropic.types.Message:
    """
    带重试机制的 API 调用封装，处理常见错误。
    """
    max_retries = 3
    retry_delay = 2
    for attempt in range(max_retries):
        try:
            kwargs = {
                "model": self.model,
                "max_tokens": self.max_tokens,
                "temperature": self.temperature,
                "system": self.system_prompt,
                "messages": messages,
                "stream": stream
            }
            if tools:
                kwargs["tools"] = tools
            if stream:
                return self.client.messages.create(**kwargs)
            else:
                return self.client.messages.create(**kwargs)
        except anthropic.RateLimitError:
            if attempt < max_retries - 1:
                print(f"\n[系统] 请求频率过高，{retry_delay}秒后重试...")
                time.sleep(retry_delay)
                retry_delay *= 2
            else:
                raise
        except anthropic.APIStatusError as e:
            if e.status_code >= 500 and attempt < max_retries - 1:
                print(f"\n[系统] 服务暂时不可用，{retry_delay}秒后重试...")
                time.sleep(retry_delay)
                retry_delay *= 2
            else:
                raise

def _execute_tool(self, tool_name: str, tool_input: dict) -> str:
    """
    执行工具调用并返回结果字符串。
    实际项目中，这里应连接真实 API 或数据库。
    """
    if tool_name == "get_weather":
        city = tool_input.get("city", "未知城市")
        # 模拟天气数据
        return f"{city}当前天气：晴朗，气温 22°C，湿度 45%，风速 3m/s。"
    else:
        return f"未知工具：{tool_name}"

def _process_tool_calls(self, content_blocks: List[Dict]) -> str:
    """
    处理所有 tool_use 块，执行工具并返回格式化结果。
    """
    tool_results = []
    for block in content_blocks:
        if block.get("type") == "tool_use":
            tool_name = block["name"]
            tool_input = block["input"]
            tool_id = block["id"]
            result = self._execute_tool(tool_name, tool_input)
            tool_results.append({
                "tool_use_id": tool_id,
                "content": result,
                "type": "tool_result"
            })
    return tool_results

def chat(self, user_message: str) -> str:
    """
    主对话接口：接收用户消息，返回助手回复。
    自动处理工具调用循环。
    """
    # 添加用户消息到历史
    self.conversation.append({"role": "user", "content": user_message})
    self._manage_history()

    # 构建 API 消息列表（直接使用历史）
    messages = self.conversation.copy()

    # 第一次调用，可能返回文本或工具调用
    response = self._call_api(messages, tools=self.tools, stream=False)

    # 检查是否有 tool_use 块
    content_blocks = response.content
    has_tool_use = any(block.type == "tool_use" for block in content_blocks)

    if has_tool_use:
        # 将 assistant 的原始响应（包含 tool_use）加入历史
        assistant_content = []
        for block in content_blocks:
            if block.type == "text":
                assistant_content.append({"type": "text", "text": block.text})
            elif block.type == "tool_use":
                assistant_content.append({
                    "type": "tool_use",
                    "id": block.id,
                    "name": block.name,
                    "input": block.input
                })
        self.conversation.append({"role": "assistant", "content": assistant_content})

        # 执行工具并构造 tool_result 消息
        tool_results = self._process_tool_calls(content_blocks)
        self.conversation.append({"role": "user", "content": tool_results})

        # 再次调用模型，获取基于工具结果的最终回复
        final_response = self._call_api(self.conversation.copy(), tools=None, stream=False)
        # 提取文本内容
        final_text = "".join(
            block.text for block in final_response.content if block.type == "text"
        )
        # 将最终回答内容加入历史
        self.conversation.append({"role": "assistant", "content": final_text})
        return final_text
    else:
        # 直接文本回复
        reply_text = "".join(
            block.text for block in content_blocks if block.type == "text"
        )
        self.conversation.append({"role": "assistant", "content": reply_text})
        return reply_text

def chat_stream(self, user_message: str):
    """
    流式对话接口：逐字输出助手回复，支持工具调用（但工具调用时回退为非流式）。
    """
    self.conversation.append({"role": "user", "content": user_message})
    self._manage_history()

    messages = self.conversation.copy()
    stream = self._call_api(messages, tools=self.tools, stream=True)

    collected_content = []
    has_tool_use = False
    tool_use_blocks = []

    print("Assistant: ", end="", flush=True)
    with stream as stream_events:
        for event in stream_events:
            # 不同类型事件：content_block_start, content_block_delta, content_block_stop等
            if event.type == "content_block_start":
                if event.content_block.type == "tool_use":
                    has_tool_use = True
                    tool_use_blocks.append({
                        "id": event.content_block.id,
                        "name": event.content_block.name,
                        "input": {}
                    })
            elif event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    print(event.delta.text, end="", flush=True)
                elif event.delta.type == "input_json_delta" and has_tool_use:
                    # 累积工具调用的 JSON 输入
                    if tool_use_blocks:
                        tool_use_blocks[-1]["input"] = json.loads(
                            event.delta.partial_json
                        ) if event.delta.partial_json else {}
            elif event.type == "content_block_stop":
                pass
    print()  # 换行

    if has_tool_use:
        # 流式模式中检测到工具调用，需重新用非流式处理完整工具流程
        print("[系统] 检测到工具调用，正在执行...")
        # 注意：流式响应中的 assistant 消息仍未加入历史，这里需要重建
        # 简便做法：从历史中移除最后的 user 消息，重新用非流式 chat 方法处理
        self.conversation.pop()  # 移除刚添加的 user 消息
        return self.chat(user_message)
    else:
        # 文本回复已输出，需将完整内容加入历史
        # 由于流式没有直接完整文本，我们记录已打印文本。
        # 这里简化处理：重新用非流式获取一次以得到完整消息对象加入历史
        self.conversation.pop()  # 移除 user 消息
        return self.chat(user_message)

def reset(self):
    """清空对话历史"""
    self.conversation = []

def main():
assistant = ClaudeAssistant(
system_prompt=(
"你是一个乐于助人的助手，可以使用工具查询天气。"
"请用简洁友好的中文回答用户。"
)
)

print("