Claude API 历史对话回溯技术解析与实现方案

1次阅读

共计 2830 个字符，预计需要花费 8 分钟才能阅读完成。

开发者在使用 Claude API 构建对话应用时，经常面临一个棘手问题：默认情况下，Claude 不会自动保存对话历史。这意味着每次 API 调用都是独立的，缺乏上下文连续性，严重影响用户体验。尤其是在多轮对话场景中，无法回溯历史对话会导致 AI 助手显得 ” 健忘 ”，无法进行有深度的持续交流。

Claude API 通过 conversation_id 字段实现对话连续性，该 ID 在首次对话时由服务端生成
后续请求携带相同 conversation_id 时，Claude 会自动关联上下文，最长支持 20 轮对话记忆
会话 ID 有效期通常为 24 小时，超时后需要重新建立会话

优势：集中管理，多设备同步访问
劣势：需要额外开发 API 接口，存在网络延迟
适用场景：多终端访问的企业级应用

优势：响应速度快，实现简单
劣势：数据仅限单设备访问
适用场景：单机版工具或移动端应用

import json
from typing import Dict, Optional
import os
from datetime import datetime, timedelta
import hashlib
from claude_api import Client  # 假设使用第三方 Claude 客户端库

class ConversationManager:
    """Claude 对话历史管理器"""

    def __init__(self, cache_dir: str = "./conversation_cache"):
        self.client = Client()
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def _get_cache_path(self, conversation_id: str) -> str:
        """生成加密缓存文件路径"""
        hash_id = hashlib.sha256(conversation_id.encode()).hexdigest()
        return os.path.join(self.cache_dir, f"{hash_id}.json")

    def save_conversation(
        self, 
        conversation_id: str, 
        messages: Dict[str, str],
        expires: int = 86400  # 默认 24 小时过期
    ) -> None:
        """保存对话历史到本地缓存"""
        cache_data = {
            "conversation_id": conversation_id,
            "messages": messages,
            "expires_at": (datetime.now() + timedelta(seconds=expires)).isoformat()}

        with open(self._get_cache_path(conversation_id), 'w') as f:
            json.dump(cache_data, f, indent=2)

    def load_conversation(self, conversation_id: str) -> Optional[Dict[str, str]]:
        """从缓存加载对话历史"""
        try:
            cache_path = self._get_cache_path(conversation_id)
            if not os.path.exists(cache_path):
                return None

            with open(cache_path, 'r') as f:
                data = json.load(f)

            if datetime.fromisoformat(data["expires_at"]) < datetime.now():
                os.remove(cache_path)
                return None

            return data["messages"]
        except Exception as e:
            print(f"加载对话缓存失败: {e}")
            return None

    def send_message(
        self, 
        prompt: str, 
        conversation_id: str = None,
        max_retries: int = 3
    ) -> Dict:
        """发送消息到 Claude API（带重试机制）"""
        history = self.load_conversation(conversation_id) if conversation_id else None

        for attempt in range(max_retries):
            try:
                response = self.client.send_message(
                    prompt=prompt,
                    conversation_id=conversation_id,
                    history=history
                )

                # 保存更新后的对话历史
                new_history = {
                    "user": prompt,
                    "assistant": response["content"]
                }
                self.save_conversation(response["conversation_id"], 
                    new_history
                )

                return {"content": response["content"],
                    "conversation_id": response["conversation_id"]
                }

            except Exception as e:
                if attempt == max_retries - 1:
                    raise RuntimeError(f"API 请求失败: {e}")

                time.sleep(1 * (attempt + 1))  # 指数退避

# 使用示例
if __name__ == "__main__":
    manager = ConversationManager()

    # 新对话
    response = manager.send_message("你好，Claude")
    print(response["content"])

    # 继续对话
    response = manager.send_message("我刚才问了什么？", response["conversation_id"])
    print(response["content"])