Telegram Bot 上下文管理重构技术设计文档

文档信息

项目内容
文档标题Telegram Bot 上下文管理重构技术设计
版本V2.2
作者Claude Agent
创建日期2025-12-10
最后更新2025-12-10
状态待评审

修订记录

版本日期作者变更内容
V1.02025-12-10Claude Agent初始版本 - 基于 claude-code-sdk 的手动上下文管理方案
V2.02025-12-10Claude Agent重大更新 - 迁移到 claude-agent-sdk，使用内置 automatic compaction
V2.12025-12-10Claude Agent简化持久化 - Session 恢复即历史恢复，移除本地对话历史存储
V2.22025-12-10Claude Agent修正 Session 存储位置 - Session 数据存储在本地 CLI 目录，非服务端

1. 背景与问题

1.1 问题描述

Telegram Bot 在长时间对话后出现“抽风”现象：

  • 新发送的消息不被响应
  • Bot 回复内容与上下文无关或完全混乱
  • API 请求超时或失败

1.2 根本原因分析

通过代码审查发现，问题根源是上下文管理机制的设计缺陷：

问题1: 提示词爆炸式增长

位置: src/claude_agent/telegram/claude_adapter.py _build_full_prompt() 方法

每次请求都将 CLAUDE.md + 全部历史消息 + 新消息 拼接成一个巨大字符串
没有任何 Token 限制检查

问题2: 历史记录污染

位置: src/claude_agent/core/agent.py process_user_input() 方法

完整的 full_prompt (包含系统提示和所有历史) 被原样存入 conversation_history
导致历史记录指数级膨胀

问题3: 粗暴的截断机制

位置: src/claude_agent/telegram/claude_adapter.py 第758-773行

if len(full_prompt) > 1000000:  # 1MB 才触发！
    full_prompt = full_prompt[:20000]  # 字符截断，破坏上下文结构

1.3 SDK 现状分析

当前使用: claude-code-sdk (已废弃)

  • 无状态的 query() 函数
  • 无内置上下文管理
  • 无自动压缩功能
  • 已停止维护

应该迁移到: claude-agent-sdk (新版)

  • 内置 Automatic Compaction - 自动上下文压缩
  • ClaudeSDKClient 支持多轮对话
  • 内置 Session 管理
  • 活跃维护，最新版本 0.1.13 (2025-12-05)

2. 设计目标

2.1 核心目标

  1. 迁移到 claude-agent-sdk - 使用官方推荐的新 SDK
  2. 利用内置 Automatic Compaction - 不再手动管理上下文压缩
  3. 使用 ClaudeSDKClient - 支持真正的多轮对话
  4. 保持功能完整 - 持久化、多平台支持不受影响

2.2 性能指标

指标目标值
最大支持对话轮数无限制 (SDK 自动压缩)
单条消息响应时间< 5秒
上下文管理SDK 内置，自动处理
内存占用（每会话）< 1MB

2.3 约束条件

  • 保持与 SSHOUT、CLI 等其他平台的兼容性
  • 保持持久化存储功能（重启后恢复对话）
  • 不改变现有配置文件格式
  • Python >= 3.10 (新 SDK 要求)

3. 方案对比

3.1 SDK 对比

特性claude-code-sdk (旧)claude-agent-sdk (新)
维护状态❌ 已废弃✅ 活跃维护
最新版本0.0.25 (2025-09-29)0.1.13 (2025-12-05)
Context 管理❌ 无，需手动实现✅ 内置 Automatic Compaction
多轮对话❌ 每次 query() 独立✅ ClaudeSDKClient 支持
Session 管理❌ 无✅ 内置
工具生态基础丰富（文件、代码执行、MCP）
Hooks有限完整（PreToolUse, PostToolUse 等）

3.2 方案对比

方案描述优点缺点
V1.0 手动管理自己实现 Token 计数、截断完全控制复杂、易出错、重复造轮子
V2.0 SDK 内置使用 claude-agent-sdk 的 automatic compaction简单、可靠、官方维护需要迁移代码

选择: V2.0 - 迁移到 claude-agent-sdk

4. 架构设计

4.1 当前架构 (有问题)

flowchart TD
    subgraph 当前流程
        A[用户消息] --> B[_build_full_prompt]
        B --> C["拼接: CLAUDE.md + 全部历史 + 消息"]
        C --> D["query() - 无状态调用"]
        D --> E["存入 conversation_history<br/>(污染的数据)"]
        E --> F["下次请求: 历史更大"]
        F --> G["最终: 超出限制, Bot 抽风"]
    end

    style C fill:#ff6b6b
    style E fill:#ff6b6b
    style G fill:#ff6b6b

4.2 目标架构 (基于 claude-agent-sdk)

flowchart TD
    subgraph 新流程
        A[用户消息] --> B[ClaudeSDKClient]
        B --> C{上下文接近限制?}
        C -->|是| D["SDK 自动触发 Compaction<br/>(内置功能)"]
        D --> E[生成 Summary]
        E --> F[继续对话]
        C -->|否| F
        F --> G["SDK 内部管理历史"]
        G --> H[响应用户]
        H --> I["持久化 Session 状态"]
    end

    style D fill:#51cf66
    style E fill:#51cf66
    style G fill:#51cf66

4.3 组件架构

classDiagram
    class ClaudeAgentAdapter {
        <<重构>>
        -_clients: Dict[str, ClaudeSDKClient]
        -_options: ClaudeAgentOptions
        +get_or_create_client(chat_id): ClaudeSDKClient
        +send_message(chat_id, message): AsyncGenerator
        +close_client(chat_id)
        -_build_options(): ClaudeAgentOptions
    }

    class ClaudeSDKClient {
        <<SDK 内置>>
        +query(prompt): void
        +receive_response(): AsyncGenerator
        +close(): void
        -automatic_compaction: 内置
        -session_management: 内置
    }

    class ClaudeAgentOptions {
        <<SDK 内置>>
        +system_prompt: str
        +max_turns: int
        +allowed_tools: List[str]
        +permission_mode: str
        +cwd: Path
        +mcp_servers: Dict
        +hooks: Dict
    }

    class SessionPersistence {
        <<新增>>
        +save_session(chat_id, client_state)
        +load_session(chat_id): Optional[state]
        +clear_session(chat_id)
    }

    ClaudeAgentAdapter --> ClaudeSDKClient : manages
    ClaudeAgentAdapter --> ClaudeAgentOptions : configures
    ClaudeAgentAdapter --> SessionPersistence : uses

5. 详细设计

5.1 SDK 迁移

5.1.1 依赖更新

requirements.txt 修改:

- claude-code-sdk>=0.0.20
+ claude-agent-sdk>=0.1.13

5.1.2 导入修改

agent.py 修改:

# 旧代码
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions, query

# 新代码
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, query
from claude_agent_sdk.types import AssistantMessage, SystemMessage, ResultMessage

5.2 ClaudeAgentAdapter 重构

5.2.1 初始化

class ClaudeAgentAdapter:
    """重构后的 Claude Agent 适配器 - 基于 claude-agent-sdk"""

    def __init__(self, config: Dict, persistence: PersistenceManager):
        self.config = config
        self.persistence = persistence
        self._clients: Dict[str, ClaudeSDKClient] = {}
        self._claude_md_content = self._load_claude_md()

        # 构建基础配置
        self._base_options = ClaudeAgentOptions(
            system_prompt=self._claude_md_content,
            max_turns=100,  # SDK 会自动 compact
            permission_mode='bypassPermissions',
            allowed_tools=['Read', 'Write', 'Bash'],
        )

5.2.2 客户端管理

async def get_or_create_client(self, chat_id: str) -> ClaudeSDKClient:
    """获取或创建指定聊天的 SDK 客户端"""
    chat_key = str(chat_id)

    if chat_key not in self._clients:
        # 尝试恢复 session
        saved_state = self.persistence.load_session_state(chat_key)

        options = ClaudeAgentOptions(
            system_prompt=self._build_system_prompt(chat_id),
            max_turns=100,
            permission_mode='bypassPermissions',
            # 如果有保存的 session，可以通过 resume 恢复
            resume=saved_state.get('session_id') if saved_state else None,
        )

        client = ClaudeSDKClient(options=options)
        await client.__aenter__()  # 初始化客户端
        self._clients[chat_key] = client

        logger.info(f"为聊天 {chat_id} 创建新的 SDK 客户端")

    return self._clients[chat_key]

5.2.3 消息处理 (利用 SDK 内置功能)

async def create_streaming_response(
    self,
    chat_id: Union[int, str],
    message: str,
    user_info: Optional[Dict] = None
) -> AsyncGenerator[str, None]:
    """
    创建流式响应 - 使用 claude-agent-sdk

    SDK 自动处理:
    - 上下文管理
    - Automatic Compaction (上下文压缩)
    - Session 管理
    """
    chat_key = str(chat_id)

    try:
        client = await self.get_or_create_client(chat_key)

        # 构建用户消息（包含用户信息）
        formatted_message = self._format_user_message(message, user_info)

        # 发送消息
        await client.query(formatted_message)

        # 接收流式响应
        full_response = ""
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if hasattr(block, 'text'):
                        chunk = block.text
                        full_response += chunk
                        yield chunk

        # 保存 session 状态用于持久化
        await self._save_session_state(chat_key, client)

        logger.info(f"聊天 {chat_id} 响应完成，长度: {len(full_response)}")

    except Exception as e:
        logger.error(f"流式响应生成错误: {e}")
        yield f"抱歉，处理请求时出现错误: {str(e)}"

5.3 Session 持久化（修正方案）

5.3.1 Session 存储机制（修正）

⚠️ 重要修正：Session 数据存储在本地 CLI 目录，而非服务端。

claude-agent-sdk 的 session 机制：

  • SDK 基于 Claude Code CLI 构建
  • Session 数据存储在本地 ~/.claude/sessions/ 目录
  • resume 参数从本地 session 文件恢复对话
  • 如果 session 文件丢失，历史对话将无法恢复

5.3.2 部署考虑

部署场景Session 文件位置风险
本地开发~/.claude/sessions/低 - 文件持久存在
Docker 容器容器内 /root/.claude/sessions/高 - 容器重启丢失
KubernetesPod 内临时存储高 - Pod 重建丢失
服务器部署服务器 ~/.claude/sessions/中 - 需确保目录持久化

5.3.3 持久化策略（双重保险）

由于 session 文件可能丢失，建议采用双重持久化策略：

class SessionPersistence:
    """Session 持久化 - 双重保险策略"""

    def __init__(self, storage_dir: Path, claude_sessions_dir: Path = None):
        self.storage_dir = storage_dir
        self.sessions_file = storage_dir / "sessions.json"
        # CLI session 目录（默认 ~/.claude/sessions/）
        self.claude_sessions_dir = claude_sessions_dir or Path.home() / ".claude" / "sessions"

    def save_session_id(self, chat_id: str, session_id: str) -> bool:
        """保存 session_id 映射"""
        try:
            sessions = self._load_sessions()
            sessions[chat_id] = {
                "session_id": session_id,
                "last_updated": int(time.time())
            }
            return self._save_sessions(sessions)
        except Exception as e:
            logger.error(f"保存 session_id 失败: {e}")
            return False

    def load_session_id(self, chat_id: str) -> Optional[str]:
        """加载 session_id（带有效性检查）"""
        sessions = self._load_sessions()
        data = sessions.get(chat_id)
        if not data:
            return None

        session_id = data.get("session_id")

        # 检查 CLI session 文件是否存在
        if session_id and not self._session_file_exists(session_id):
            logger.warning(f"Session 文件不存在: {session_id}，将创建新 session")
            return None

        return session_id

    def _session_file_exists(self, session_id: str) -> bool:
        """检查 CLI session 文件是否存在"""
        # Session 文件通常在 ~/.claude/sessions/{session_id}/
        session_path = self.claude_sessions_dir / session_id
        return session_path.exists()

    def clear_session(self, chat_id: str) -> bool:
        """清除 session"""
        try:
            sessions = self._load_sessions()
            if chat_id in sessions:
                del sessions[chat_id]
                return self._save_sessions(sessions)
            return True
        except Exception as e:
            logger.error(f"清除 session 失败: {e}")
            return False

5.3.4 恢复流程（带容错）

async def get_or_create_client(self, chat_id: str) -> ClaudeSDKClient:
    """获取或创建 SDK 客户端（带 session 有效性检查）"""
    chat_key = str(chat_id)

    if chat_key not in self._clients:
        # 尝试恢复 session（会检查文件是否存在）
        session_id = self.persistence.load_session_id(chat_key)

        options = ClaudeAgentOptions(
            system_prompt=self._claude_md_content,
            # 如果 session 文件存在才尝试恢复
            resume=session_id,
        )

        async with ClaudeSDKClient(options=options) as client:
            self._clients[chat_key] = client

            if session_id:
                logger.info(f"聊天 {chat_id} 恢复 session: {session_id}")
            else:
                logger.info(f"聊天 {chat_id} 创建新 session")

    return self._clients[chat_key]

5.3.5 Docker/K8s 部署建议

# Docker Compose 示例 - 挂载 session 目录
services:
  telegram-bot:
    image: your-bot-image
    volumes:
      - ./data/claude-sessions:/root/.claude/sessions  # 持久化 session
      - ./data/storage:/app/data/storage               # 持久化 session_id 映射

5.3.6 存储文件说明

文件位置内容说明
sessions.jsondata/storage/*/chat_id → session_id 映射我们管理
Session 文件~/.claude/sessions/对话历史、上下文CLI 管理
conversations.jsondata/storage/*/废弃不再需要
agents.jsondata/storage/*/废弃SDK 管理

5.4 Automatic Compaction 工作原理

Claude Agent SDK 的 automatic compaction 是内置功能，无需手动配置：

sequenceDiagram
    participant User as 用户
    participant Adapter as ClaudeAgentAdapter
    participant SDK as ClaudeSDKClient
    participant Claude as Claude API

    User->>Adapter: 发送消息
    Adapter->>SDK: client.query(message)

    SDK->>SDK: 检查上下文大小

    alt 上下文接近限制
        SDK->>Claude: 请求生成 Summary
        Claude-->>SDK: 返回压缩后的 Summary
        SDK->>SDK: 替换旧历史为 Summary
        Note over SDK: Automatic Compaction 完成
    end

    SDK->>Claude: 发送请求（含压缩后上下文）
    Claude-->>SDK: 流式响应
    SDK-->>Adapter: 转发响应
    Adapter-->>User: 显示响应

关键点：

  • SDK 自动监控上下文大小
  • 接近限制时自动触发压缩
  • 生成 Summary 保留关键信息
  • 对应用层透明，无需额外代码

6. 数据流图

6.1 完整消息处理流程

sequenceDiagram
    participant TG as Telegram
    participant Bot as TelegramBot
    participant Adapter as ClaudeAgentAdapter
    participant SDK as ClaudeSDKClient
    participant Persist as SessionPersistence

    TG->>Bot: 用户消息
    Bot->>Adapter: handle_message(chat_id, text)

    Adapter->>Persist: load_session_state(chat_id)
    Persist-->>Adapter: session_state (或 None)

    Adapter->>SDK: get_or_create_client(chat_id, session_state)

    Adapter->>SDK: client.query(message)
    SDK->>SDK: [内部] Automatic Compaction (如需要)

    loop 流式响应
        SDK-->>Adapter: message chunk
        Adapter-->>TG: 更新消息
    end

    Adapter->>Persist: save_session_state(chat_id, client_state)
    Persist-->>Adapter: success

7. 配置更新

7.1 新增配置项

在 configs/default.toml 中更新：

# Agent SDK 配置 (替换原有 agent 配置)
[agent]
# 思考模式: interactive 或 yolo
default_mode = "interactive"
# Claude 模型配置 (SDK 默认使用最新模型)
model = "claude-sonnet-4-5-20250929"
# API 超时时间
api_timeout = 60
# 最大对话轮数 (SDK 会自动 compact，可以设置较大值)
max_turns = 100
# 权限模式: default, acceptEdits, plan, bypassPermissions
permission_mode = "bypassPermissions"

# Session 持久化配置
[agent.session]
# 是否启用 session 持久化
enabled = true
# Session 过期时间 (秒)，0 表示永不过期
expire_seconds = 86400

8. 核心代码修改点

8.1 修改清单

文件修改类型修改内容
requirements.txt修改claude-code-sdk → claude-agent-sdk
src/claude_agent/core/agent.py重构使用 ClaudeSDKClient 替代 query()
src/claude_agent/telegram/claude_adapter.py重构移除手动上下文管理，使用 SDK 内置功能
src/claude_agent/storage/persistence.py简化只保留 session_id 存储，删除对话历史存储
configs/default.toml修改更新配置项

8.2 删除的代码

以下代码将被删除（不再需要手动管理）：

文件删除内容原因
agent.pyclean_conversation_history()SDK 自动管理
agent.pytrim_conversation_history()SDK 自动管理
agent.pyconversation_history 列表服务端存储
claude_adapter.py_build_full_prompt() 中的历史拼接SDK 自动管理
claude_adapter.py1MB 截断逻辑 (758-773行)SDK 自动 compact
claude_adapter.py_merge_conversation_history()SDK 内部管理
persistence.pysave_conversation_history()服务端存储
persistence.pyload_conversation_history()服务端存储

8.3 废弃的存储文件

文件状态说明
data/storage/*/conversations.json废弃对话历史由服务端存储
data/storage/*/agents.json废弃Agent 状态由 SDK 管理
data/storage/*/sessions.json新增只存 session_id 映射

9. 风险评估

风险可能性影响缓解措施
SDK 升级导致 API 变化中中锁定版本号，逐步升级
Session 文件丢失中高挂载持久化卷，检查文件存在性
Docker/K8s 重启丢失高高必须挂载 ~/.claude/sessions/ 目录
Session 恢复失败中低失败时创建新 session，优雅降级
Compaction 质量不稳定低中SDK 官方维护，持续优化
Python 版本要求提高低中项目已使用 Python 3.10+

10. 测试计划

10.1 单元测试

测试项测试内容预期结果
SDK 导入导入 claude-agent-sdk成功导入
客户端创建创建 ClaudeSDKClient成功创建
基本对话发送消息并接收响应正常响应
Session 持久化保存和加载 session数据一致

10.2 集成测试

测试场景步骤预期结果
长对话测试连续发送 100+ 条消息Bot 持续正常响应 (SDK 自动 compact)
重启恢复测试对话后重启 BotSession 正确恢复
并发测试多个群组同时对话上下文不串台
Compaction 测试触发自动压缩对话继续，信息保留

10.3 验收标准

  • [ ] 长对话不再出现“抽风”现象
  • [ ] SDK 自动 compact 正常工作
  • [ ] Session 持久化和恢复正常
  • [ ] 性能无明显下降

11. 快速参考表 (For Code Review)

核心组件/配置实现状态文件路径核心功能关键代码位置
依赖更新
requirements.txt待修改requirements.txtSDK 依赖claude-agent-sdk>=0.1.13
核心组件
ClaudeAgentAdapter待重构src/claude_agent/telegram/claude_adapter.py使用 ClaudeSDKClientget_or_create_client()
AgentCore待重构src/claude_agent/core/agent.py简化，移除手动管理移除 conversation_history
SessionPersistence待简化src/claude_agent/storage/persistence.py只存 session_idsave_session_id()
配置文件
default.toml待修改configs/default.toml更新 SDK 配置[agent] 节
待删除代码
手动历史管理待删除agent.pyconversation_history 列表N/A
手动截断待删除claude_adapter.py1MB 截断逻辑N/A
历史合并待删除claude_adapter.py_merge_conversation_history()N/A
对话存储待删除persistence.pysave/load_conversation_history()N/A
待废弃文件
conversations.json废弃data/storage/*/对话历史 → 服务端存储N/A
agents.json废弃data/storage/*/Agent 状态 → SDK 管理N/A

12. 实施计划

Phase 1: 环境准备

  1. 升级 Python 版本确认 (>= 3.10)
  2. 更新 requirements.txt
  3. 安装 claude-agent-sdk

Phase 2: 核心重构

  1. 重构 AgentCore - 使用 ClaudeSDKClient
  2. 重构 ClaudeAgentAdapter - 移除手动上下文管理
  3. 实现 SessionPersistence

Phase 3: 测试验证

  1. 运行单元测试
  2. 进行长对话压力测试
  3. 验证 automatic compaction

Phase 4: 清理

  1. 删除废弃代码
  2. 更新文档
  3. 更新 README

13. Review Checklist

  • [ ] SDK 依赖正确更新为 claude-agent-sdk
  • [ ] ClaudeSDKClient 正确初始化和管理
  • [ ] 移除了所有手动上下文管理代码
  • [ ] Session 持久化正常工作
  • [ ] 长对话测试通过（100+ 轮无抽风）
  • [ ] Automatic compaction 正常触发
  • [ ] 向后兼容（现有配置、数据）
  • [ ] 单元测试覆盖率达到 80%

附录 A: 参考资料

附录 B: 与 V1.0 方案对比

对比项V1.0 手动管理V2.2 SDK 内置
SDKclaude-code-sdk (废弃)claude-agent-sdk (推荐)
上下文管理手动实现 TokenEstimator, ContextBuilderSDK 内置 Automatic Compaction
压缩触发需自己实现四重条件检测SDK 自动检测和触发
Session无内置 Session 管理
对话历史存储本地 conversations.json本地 CLI 目录 (~/.claude/sessions/)
Session 目录可配置N/A不可配置 (CLI 硬编码)
本地持久化存储完整对话历史只存 session_id 映射
代码量新增 ~500 行删除 ~300 行，简化架构
维护成本高（需要跟进 Claude 变化）低（官方维护）
可靠性依赖自己实现质量官方测试保证
部署考虑无特殊要求需挂载 session 目录 (Docker/K8s)

结论: V2.2 方案更简单、更可靠、维护成本更低。但需注意 Session 存储在本地 CLI 目录，Docker/K8s 部署时必须挂载持久化卷。