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

文档信息

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

变更记录

版本日期作者变更内容
V1.02025-12-10Claude Agent初始版本

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
导致历史记录指数级膨胀：
- 第1条: CLAUDE.md + 消息1
- 第2条: CLAUDE.md + (CLAUDE.md + 消息1) + 消息2
- 第N条: 指数级增长

问题3: 粗暴的截断机制

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

if len(full_prompt) > 1000000:  # 1MB 才触发！
    full_prompt = full_prompt[:20000]  # 字符截断，破坏上下文结构
  • 1MB 阈值过高（Claude 200K token 限制早已超出）
  • 字符截断破坏 JSON、代码块、对话结构

1.3 Claude Code SDK 特性

经研究确认，Claude Code SDK 的 query() 函数是无状态的：

  • 每次调用独立，不维护会话上下文
  • 需要应用层自行管理对话历史
  • 这是正确的设计，问题在于项目的实现方式

2. 设计目标

2.1 核心目标

  1. 修复上下文膨胀问题 - 防止历史记录指数级增长
  2. 实现智能截断 - 基于 Token 而非字符进行上下文管理
  3. 保持功能完整 - 持久化、多平台支持不受影响

2.2 性能指标

指标目标值
最大支持对话轮数50+ 轮（当前配置）
单条消息响应时间< 5秒
上下文 Token 利用率> 80%
内存占用（每会话）< 1MB

2.3 约束条件

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

3. 架构设计

3.1 当前架构问题

flowchart TD
    subgraph 当前流程-有问题
        A[用户消息] --> B[_build_full_prompt]
        B --> C["拼接: CLAUDE.md + 全部历史 + 消息"]
        C --> D[传递给 AgentCore]
        D --> E["存入 conversation_history<br/>(包含完整拼接内容)"]
        E --> F[query SDK]
        F --> G[响应也存入历史]
        G --> H["下次请求时历史更大<br/>恶性循环"]
    end

    style C fill:#ff6b6b
    style E fill:#ff6b6b
    style H fill:#ff6b6b

3.2 目标架构

flowchart TD
    subgraph 新流程-修复后
        A[用户消息] --> B[ContextBuilder]
        B --> C["分离: 系统提示 / 历史 / 当前消息"]
        C --> D[TokenCounter]
        D --> E{超出限制?}
        E -->|是| F[智能截断-保留最近]
        E -->|否| G[构建最终 Prompt]
        F --> G
        G --> H[query SDK]
        H --> I["仅存储原始消息<br/>(不含系统提示)"]
        I --> J[持久化存储]
    end

    style C fill:#51cf66
    style F fill:#51cf66
    style I fill:#51cf66

3.3 组件职责重新划分

classDiagram
    class ContextBuilder {
        <<新增>>
        +system_prompt: str
        +max_tokens: int
        +build_prompt(history, message): str
        +estimate_tokens(text): int
        -_truncate_history(history, budget): List
    }

    class MessageStore {
        <<重构>>
        +add_message(role, content)
        +get_recent(limit): List
        +get_token_count(): int
        -_messages: deque
        -_max_messages: int
    }

    class ClaudeAgentAdapter {
        <<修改>>
        -context_builder: ContextBuilder
        -message_store: Dict[str, MessageStore]
        +create_streaming_response(chat_id, message)
        -_save_raw_message(chat_id, role, content)
    }

    class AgentCore {
        <<修改>>
        +conversation_history: List
        +add_message(role, content)
        +get_context_for_prompt(): List
        -_ensure_history_clean()
    }

    ContextBuilder --> MessageStore : uses
    ClaudeAgentAdapter --> ContextBuilder : uses
    ClaudeAgentAdapter --> AgentCore : manages

4. 详细设计

4.1 Token 计数器

目的: 准确估算文本的 Token 数量，用于上下文管理

实现方案: 使用简化的中英文混合估算（避免引入 tiktoken 依赖）

flowchart LR
    A[输入文本] --> B{语言检测}
    B -->|英文为主| C["字符数 / 4"]
    B -->|中文为主| D["字符数 / 1.5"]
    B -->|混合| E["加权计算"]
    C --> F[Token 估算值]
    D --> F
    E --> F

设计接口:

classDiagram
    class TokenEstimator {
        <<utility>>
        +estimate(text: str): int
        +estimate_messages(messages: List): int
        +CHARS_PER_TOKEN_EN: float = 4.0
        +CHARS_PER_TOKEN_ZH: float = 1.5
        +MAX_CONTEXT_TOKENS: int = 150000
    }

4.2 上下文构建器

目的: 分离系统提示、历史消息、当前消息的组装逻辑

核心原则:

  1. 系统提示只设置一次，不在每次请求时重复包含
  2. 历史消息只存储原始内容，不存储拼接后的 prompt
  3. 智能截断从最老的消息开始，保留最近的上下文
sequenceDiagram
    participant User as 用户消息
    participant CB as ContextBuilder
    participant TE as TokenEstimator
    participant Store as MessageStore
    participant SDK as Claude SDK

    User->>CB: build_prompt(message)
    CB->>Store: get_recent_messages()
    Store-->>CB: [msg1, msg2, ..., msgN]
    CB->>TE: estimate_tokens(system + history + message)
    TE-->>CB: total_tokens

    alt tokens > MAX_LIMIT
        CB->>CB: truncate_history(keep_recent)
        CB->>TE: re-estimate
    end

    CB->>SDK: query(final_prompt)
    SDK-->>CB: response
    CB->>Store: add_message("user", original_message)
    CB->>Store: add_message("assistant", response)

4.3 消息存储重构

当前问题: conversation_history 存储了拼接后的完整 prompt

修复方案: 只存储原始消息内容

flowchart TD
    subgraph 消息存储结构
        A[MessageStore]
        A --> B["messages: deque(maxlen=100)"]
        B --> C["{'role': 'user', 'content': '原始消息', 'timestamp': 123}"]
        B --> D["{'role': 'assistant', 'content': 'AI回复', 'timestamp': 124}"]
    end

    subgraph 禁止存储
        E["❌ 系统提示"]
        F["❌ 拼接后的历史"]
        G["❌ full_prompt"]
    end

4.4 配置扩展

在 configs/default.toml 中新增上下文管理配置：

# 上下文管理配置
[context]
# 最大上下文 Token 数 (预留 50K 给响应)
max_context_tokens = 150000
# 系统提示最大 Token 数
max_system_prompt_tokens = 10000
# 每条消息最大 Token 数 (超出则截断)
max_message_tokens = 5000
# 历史消息最小保留数量 (即使超出 token 也保留)
min_history_messages = 5
# Token 估算方式: "simple" | "tiktoken"
token_estimation = "simple"

5. 核心代码修改点

5.1 修改清单

文件修改类型修改内容
src/claude_agent/utils/token_estimator.py新增Token 估算工具类
src/claude_agent/core/context_builder.py新增上下文构建器
src/claude_agent/core/agent.py修改移除污染的历史存储逻辑
src/claude_agent/telegram/claude_adapter.py修改重构 _build_full_prompt，使用新组件
configs/default.toml修改新增 [context] 配置节

5.2 关键修改详解

5.2.1 agent.py 修改

移除: 第83行的污染存储

# 当前代码 (有问题)
async def process_user_input(self, user_input: str) -> str:
    self.conversation_history.append({"role": "user", "content": user_input})  # ❌ user_input 是完整 prompt

# 修改后
async def process_user_input(self, user_input: str) -> str:
    # 不在这里存储，由 adapter 层控制
    pass

新增: 清洁的消息添加方法

def add_raw_message(self, role: str, content: str, max_length: int = 5000):
    """添加原始消息（自动截断过长内容）"""
    if len(content) > max_length:
        content = content[:max_length] + "...[截断]"
    self.conversation_history.append({
        "role": role,
        "content": content,
        "timestamp": int(time.time())
    })
    self._trim_if_needed()

5.2.2 claude_adapter.py 修改

重构: _build_full_prompt 方法

# 当前代码 (有问题)
def _build_full_prompt(self, user_context, message, conversation_history):
    prompt_parts = []
    prompt_parts.append(f"系统提示:\n{self._claude_md_content}")  # ❌ 每次都加
    prompt_parts.append(f"群聊历史对话:\n...")  # ❌ 包含污染的历史
    return "\n\n".join(prompt_parts)

# 修改后
def _build_full_prompt(self, user_context: str, message: str,
                       conversation_history: List[Dict]) -> str:
    """构建提示词（使用 ContextBuilder）"""
    return self._context_builder.build(
        system_prompt=self._claude_md_content,
        history=conversation_history,
        user_context=user_context,
        current_message=message
    )

新增: 响应后只存储原始消息

async def _handle_response(self, chat_id: str, user_message: str, response: str):
    """处理响应后的存储"""
    agent = self._agents.get(chat_id)
    if agent:
        # 只存储原始消息，不存储 full_prompt
        agent.add_raw_message("user", user_message)
        agent.add_raw_message("assistant", response)
        await self._save_agent_state(chat_id, agent)

6. 数据流图

6.1 消息处理完整流程

sequenceDiagram
    participant TG as Telegram
    participant Bot as TelegramBot
    participant Adapter as ClaudeAgentAdapter
    participant CB as ContextBuilder
    participant Agent as AgentCore
    participant SDK as Claude SDK
    participant Store as PersistenceManager

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

    Adapter->>Adapter: _get_or_create_agent(chat_id)
    Adapter->>Agent: get_conversation_history()
    Agent-->>Adapter: [原始历史消息]

    Adapter->>CB: build(system, history, message)
    CB->>CB: estimate_tokens()
    CB->>CB: truncate_if_needed()
    CB-->>Adapter: final_prompt

    Adapter->>SDK: query(final_prompt)
    SDK-->>Adapter: response_stream

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

    Adapter->>Agent: add_raw_message("user", original_text)
    Adapter->>Agent: add_raw_message("assistant", response)
    Adapter->>Store: save_agent_state()

6.2 Token 预算分配

pie title Token 预算分配 (200K 总限制)
    "系统提示 (CLAUDE.md)" : 10
    "用户上下文" : 2
    "历史消息" : 60
    "当前消息" : 8
    "预留给响应" : 20

7. 风险评估

风险可能性影响缓解措施
Token 估算不准确中低预留 20% 安全边际；支持切换到 tiktoken
历史截断丢失重要上下文中中保证最少 5 条历史；优先截断最老消息
持久化数据格式不兼容低高添加版本号；编写迁移脚本
多平台行为不一致低中统一使用 ContextBuilder

8. 测试计划

8.1 单元测试

测试项测试内容预期结果
TokenEstimator.estimate中英文混合文本误差 < 20%
ContextBuilder.build正常上下文构建包含所有必要部分
ContextBuilder.truncate超长历史截断保留最近消息
Agent.add_raw_message消息存储只存储原始内容
Agent.add_raw_message超长消息自动截断

8.2 集成测试

测试场景步骤预期结果
长对话测试连续发送 100+ 条消息Bot 持续正常响应
大消息测试发送 10KB+ 的长消息消息被截断，Bot 正常响应
重启恢复测试对话后重启 Bot历史正确恢复，无污染
并发测试多个群组同时对话上下文不串台

8.3 黑盒测试

# 1. 长对话压力测试
python tests/blackbox/test_long_conversation.py

# 2. 上下文正确性验证
python tests/blackbox/test_context_integrity.py

# 3. 持久化恢复测试
python tests/blackbox/test_persistence_recovery.py

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

核心组件/配置实现状态文件路径核心功能关键代码位置
新增组件
TokenEstimator待实现src/claude_agent/utils/token_estimator.pyToken 估算estimate() 方法
ContextBuilder待实现src/claude_agent/core/context_builder.py上下文构建build() 方法
修改组件
AgentCore待修改src/claude_agent/core/agent.py移除污染存储add_raw_message() 方法
ClaudeAgentAdapter待修改src/claude_agent/telegram/claude_adapter.py重构 prompt 构建_build_full_prompt() 方法
配置文件
default.toml待修改configs/default.toml新增 context 配置[context] 节
待删除代码
历史污染存储待删除src/claude_agent/core/agent.py第83行 append 逻辑N/A
粗暴截断逻辑待删除src/claude_agent/telegram/claude_adapter.py第758-773行N/A

10. 实施计划

Phase 1: 基础组件 (无破坏性修改)

  1. 新增 TokenEstimator 类
  2. 新增 ContextBuilder 类
  3. 编写单元测试

Phase 2: 核心重构

  1. 修改 AgentCore.add_raw_message()
  2. 重构 ClaudeAgentAdapter._build_full_prompt()
  3. 更新配置文件

Phase 3: 清理与测试

  1. 移除旧的污染代码
  2. 运行完整测试套件
  3. 进行黑盒测试验证

Phase 4: 数据迁移

  1. 编写历史数据清理脚本
  2. 清理现有被污染的 conversations.json

11. Review Checklist

  • [ ] Token 估算逻辑是否准确（误差 < 20%）
  • [ ] 上下文构建是否正确分离系统提示和历史
  • [ ] 消息存储是否只存储原始内容
  • [ ] 截断逻辑是否保留最近消息
  • [ ] 持久化数据格式是否向后兼容
  • [ ] 配置项是否合理
  • [ ] 单元测试覆盖率是否达到 80%
  • [ ] 是否有潜在的内存泄漏
  • [ ] 多平台（Telegram/SSHOUT/CLI）行为是否一致
  • [ ] 错误处理是否完善

附录 A: 参考资料