ReAct模式Agent中的工具调用历史管理与Token优化策略

Posted on

问题背景

在实现 ReAct（Reasoning + Acting）模式的 AI Agent 时，我们面临一个核心挑战：如何高效管理工具调用历史以优化 Token 使用？

ReAct 模式的标准流程

  1. 初始化：将 MCP 的 tool 列表转换为 function calling 格式，传给 LLM
  2. LLM 推理：分析请求，决定调用哪些工具及参数
  3. 执行 Tool：调用对应的 MCP tool，获取返回结果
  4. 反馈循环：将结果添加到对话历史，再次发送给 LLM
  5. 迭代判断：
    • 需要更多信息 → 继续调用工具
    • 信息充足 → 生成最终回复

核心问题

当 Agent 处理用户的下一个输入时，需要带上 history + 最新 input。那么：

history 是否有必要带上历史 tool 调用信息？

这个看似简单的问题，实际涉及到：

  • 上下文连贯性 vs Token 成本
  • 多轮推理能力 vs 内存限制
  • 准确性保证 vs 性能优化

研究方法论

根据 Anthropic 和 LangChain 的最新研究，我们找到了几种经过验证的解决方案。

方案一：Tool Result Clearing（工具结果清理）

核心思想

“一旦 tool 在消息历史深处被调用过，agent 为什么还需要再次看到原始结果？”
—— Anthropic Research

保留工具调用的元数据，但清除详细的返回结果。

实现方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
// ❌ 原始方式：保留完整历史
[
  { role: "user", content: "北京今天天气怎么样？" },
  {
    role: "assistant",
    content: null,
    tool_calls: [{
      id: "1",
      name: "get_weather",
      arguments: {city: "北京"}
    }]
  },
  {
    role: "tool",
    tool_call_id: "1",
    content: "{\"temp\": 20, \"condition\": \"sunny\", \"humidity\": 65, \"wind\": \"3m/s\", ...}" // 大量数据
  },
  { role: "assistant", content: "北京今天晴天，20°C" },
  { role: "user", content: "那明天呢？" }
]

// ✅ Tool Result Clearing：清理旧结果
[
  { role: "user", content: "北京今天天气怎么样？" },
  {
    role: "assistant",
    content: null,
    tool_calls: [{
      id: "1",
      name: "get_weather",
      arguments: {city: "北京"}
    }]
  },
  {
    role: "tool",
    tool_call_id: "1",
    content: "[Result cleared - 已处理]" // 仅保留标记
  },
  { role: "assistant", content: "北京今天晴天，20°C" },
  { role: "user", content: "那明天呢？" }
]

优势

  • ✅ 最安全：保留调用记录，LLM 知道调用了什么工具
  • ✅ 最轻量级：大幅减少 Token 占用
  • ✅ 已产品化：Claude 开发者平台已支持此功能

适用场景

  • 工具返回结果体积大（如文档检索、数据库查询）
  • 多轮对话后累积大量历史
  • 短期优化快速生效

方案二：Summarization（摘要压缩）

核心思想

对较旧的对话历史进行智能摘要，保留最近交互的完整信息。

实现策略

1. 滑动窗口 + 摘要

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class HistoryManager:
    def __init__(self, max_recent_turns=5):
        self.max_recent_turns = max_recent_turns
        self.full_history = []
        self.summary = ""

    def add_interaction(self, user_msg, assistant_msg, tool_calls):
        self.full_history.append({
            "user": user_msg,
            "assistant": assistant_msg,
            "tools": tool_calls
        })

        # 当历史超过阈值时，压缩旧历史
        if len(self.full_history) > self.max_recent_turns + 5:
            old_history = self.full_history[:-self.max_recent_turns]
            self.summary = self.summarize(old_history)
            self.full_history = self.full_history[-self.max_recent_turns:]

    def get_context_for_llm(self):
        context = []

        # 添加摘要（如果有）
        if self.summary:
            context.append({
                "role": "system",
                "content": f"Previous conversation summary: {self.summary}"
            })

        # 添加最近的完整历史
        for interaction in self.full_history:
            context.append({"role": "user", "content": interaction["user"]})
            if interaction["tools"]:
                context.append({"role": "assistant", "tool_calls": interaction["tools"]})
            context.append({"role": "assistant", "content": interaction["assistant"]})

        return context

    def summarize(self, history):
        """调用 LLM 生成摘要"""
        prompt = f"""
        总结以下对话，保留关键信息：
        - 主要决策点
        - 重要发现
        - 待解决的问题

        对话内容：
        {history}
        """
        return llm.generate(prompt)

2. 分层历史管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class LayeredHistoryManager:
    def __init__(self):
        self.recent = []      # 最近 3-5 轮：完整保留
        self.mid_term = []    # 中期历史：工具调用 + 结果摘要
        self.long_term = ""   # 早期历史：整体摘要

    def compress_to_mid_term(self, interactions):
        """压缩到中期存储"""
        compressed = []
        for interaction in interactions:
            compressed.append({
                "user_query": interaction["user"],
                "tools_used": [t["name"] for t in interaction["tools"]],
                "outcome_summary": self.summarize_tool_results(interaction["tools"])
            })
        return compressed

    def compress_to_long_term(self, mid_term_history):
        """压缩到长期存储"""
        return self.generate_overall_summary(mid_term_history)

优势

  • ✅ 保持长期上下文连贯性
  • ✅ 自动平衡 Token 使用
  • ✅ Claude Code 已采用（95% 上下文窗口时自动压缩）

适用场景

  • 长时间多轮对话
  • 需要跨多轮推理的复杂任务
  • 用户可能引用早期对话内容

方案三：Memory Blocks（草稿本/记忆块）

核心思想

Agent 主动管理重要信息，将关键内容持久化到上下文窗口之外。

实现方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class MemoryBlockSystem:
    def __init__(self):
        self.memory_blocks = {
            "facts": [],         # 已知事实
            "decisions": [],     # 关键决策
            "pending_tasks": [], # 待办事项
            "key_findings": {},  # 研究发现
            "context": {}        # 重要上下文
        }

    def save_to_memory(self, block_type, content, metadata=None):
        """Agent 主动保存信息"""
        entry = {
            "timestamp": datetime.now(),
            "content": content,
            "metadata": metadata or {}
        }

        if block_type in self.memory_blocks:
            self.memory_blocks[block_type].append(entry)

    def retrieve_relevant_memory(self, query, top_k=5):
        """检索相关记忆"""
        all_entries = []
        for block_type, entries in self.memory_blocks.items():
            for entry in entries:
                similarity = self.compute_similarity(query, entry["content"])
                all_entries.append((similarity, block_type, entry))

        # 返回最相关的记忆
        all_entries.sort(reverse=True, key=lambda x: x[0])
        return all_entries[:top_k]

    def get_memory_summary(self):
        """获取记忆摘要，注入到 LLM 上下文"""
        return {
            "total_facts": len(self.memory_blocks["facts"]),
            "pending_decisions": len(self.memory_blocks["decisions"]),
            "key_findings": list(self.memory_blocks["key_findings"].keys())
        }

使用示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# Agent 在执行任务时主动保存信息
memory = MemoryBlockSystem()

# 保存重要发现
memory.save_to_memory("key_findings", {
    "topic": "性能优化",
    "finding": "使用 Tool Result Clearing 可减少 40% Token"
})

# 保存待办任务
memory.save_to_memory("pending_tasks", "验证压缩策略在长对话中的效果")

# 在新对话中检索相关记忆
relevant_memories = memory.retrieve_relevant_memory("如何优化 Token 使用")

优势

  • ✅ 突破上下文窗口限制
  • ✅ Agent 主动管理信息
  • ✅ 适合超长对话（>200K tokens）

适用场景

  • 复杂研究任务
  • 多会话协作
  • 需要跨天/跨周的长期任务

方案四：History-Based Deletion（基于历史的删除）

核心思想

使用效用评估器判断历史消息的价值，删除低价值内容。

实现方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class UtilityBasedHistoryManager:
    def __init__(self, max_history_tokens=10000):
        self.max_tokens = max_history_tokens
        self.history = []

    def compute_utility(self, message):
        """评估消息的效用分数"""
        score = 0

        # 1. 时间衰减（越旧越不重要）
        age = datetime.now() - message["timestamp"]
        time_decay = math.exp(-age.total_seconds() / 3600)  # 1小时衰减
        score += time_decay * 0.3

        # 2. 引用频率（被后续对话引用越多越重要）
        score += message.get("reference_count", 0) * 0.3

        # 3. 包含关键信息（工具调用、决策点）
        if message.get("has_tool_call"):
            score += 0.2
        if message.get("is_decision_point"):
            score += 0.2

        return score

    def prune_history(self):
        """删除低效用的历史"""
        # 计算当前历史的总 token
        total_tokens = sum(msg["token_count"] for msg in self.history)

        if total_tokens <= self.max_tokens:
            return

        # 按效用排序
        scored_history = [(self.compute_utility(msg), msg) for msg in self.history]
        scored_history.sort(reverse=True, key=lambda x: x[0])

        # 保留高效用消息，直到达到 token 限制
        pruned_history = []
        current_tokens = 0

        for score, msg in scored_history:
            if current_tokens + msg["token_count"] <= self.max_tokens:
                pruned_history.append(msg)
                current_tokens += msg["token_count"]
            else:
                break

        # 按时间顺序重新排序
        self.history = sorted(pruned_history, key=lambda x: x["timestamp"])

优势

  • ✅ 智能删除，保留重要信息
  • ✅ 自适应 Token 管理
  • ✅ 研究验证有效

适用场景

  • 对话内容价值差异大
  • 需要精确 Token 控制
  • 长期部署的 Agent 系统

方案五：Sub-Agent 架构（子Agent）

核心思想

将复杂任务分解给专门的子 Agent，每个子 Agent 有独立的上下文窗口。

架构设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class MasterAgent:
    def __init__(self):
        self.sub_agents = {
            "researcher": ResearchAgent(),
            "coder": CodingAgent(),
            "reviewer": ReviewAgent()
        }
        self.task_results = {}

    def execute_complex_task(self, task):
        # 主 Agent 分解任务
        subtasks = self.decompose_task(task)

        results = []
        for subtask in subtasks:
            # 选择合适的子 Agent
            agent_type = self.select_agent(subtask)
            agent = self.sub_agents[agent_type]

            # 子 Agent 独立执行（有自己的上下文）
            result = agent.execute(subtask)

            # 子 Agent 返回浓缩的摘要
            summary = result.get_summary()  # 通常 1,000-2,000 tokens
            results.append(summary)

        # 主 Agent 整合结果
        return self.integrate_results(results)

class ResearchAgent:
    def __init__(self):
        self.context = []  # 独立的上下文窗口
        self.tools = ["search", "read_paper", "summarize"]

    def execute(self, research_task):
        # 子 Agent 可能使用数万个 tokens 深度探索
        for step in research_task.steps:
            result = self.use_tool(step.tool, step.params)
            self.context.append(result)

        # 但只返回浓缩摘要
        return ResearchResult(
            summary=self.generate_summary(),
            key_findings=self.extract_key_findings(),
            next_actions=self.suggest_next_actions()
        )

优势

  • ✅ 绕过单一上下文窗口限制
  • ✅ 专业化分工，提高效率
  • ✅ 主 Agent 只处理高层次摘要

适用场景

  • 极复杂任务（如完整代码库分析）
  • 需要不同专业能力的任务
  • 超长上下文需求（>100K tokens）

实践建议

短期快速实现

  1. Tool Result Clearing + 滑动窗口

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    # 保留最近 N 轮完整历史，清理旧工具结果
    MAX_RECENT_TURNS = 10

    def prepare_context(history):
        recent = history[-MAX_RECENT_TURNS:]

        for msg in recent[:-5]:  # 除了最近5轮
            if msg["role"] == "tool":
                msg["content"] = "[Result cleared]"

        return recent

  2. Token 监控

    1
    2
    3
    4
    5
    def count_tokens(messages):
        return sum(len(msg["content"]) // 4 for msg in messages)

    def should_compress(messages, threshold=8000):
        return count_tokens(messages) > threshold

中期优化

  1. 智能压缩：对大数据工具结果即时摘要
  2. 分层管理：实现 3 层历史结构
  3. 效用评估：基于引用频率和时间衰减

长期方案

  1. 专用摘要模型：微调小模型专门做摘要
  2. Memory Blocks 系统：完整的记忆管理
  3. Sub-Agent 架构：复杂任务分解

性能对比

方案 Token 节省 实现复杂度 上下文保留 适用场景
Tool Result Clearing 40-60% 低 中 日常对话
Summarization 50-70% 中 高 长对话
Memory Blocks 70-80% 高 极高 超长任务
History-Based Deletion 50-65% 中 中高 精确控制
Sub-Agent 80-90% 高 专业化 复杂任务

案例研究：Claude Code 的实践

Claude Code 采用了多层次组合策略：

  1. 自动压缩：上下文达到 95% 时触发摘要
  2. Tool Result Clearing：默认清理深度历史中的工具结果
  3. Memory Blocks：让 Agent 主动保存计划和发现
  4. Sub-Agent：复杂任务使用专门的子 Agent

这种组合策略使 Claude Code 能够处理超长对话，同时保持高效的 Token 使用。

代码示例：完整实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
class OptimizedReActAgent:
    def __init__(self, llm, tools, max_context_tokens=10000):
        self.llm = llm
        self.tools = tools
        self.max_context_tokens = max_context_tokens

        # 多策略管理器
        self.history_manager = LayeredHistoryManager()
        self.memory_blocks = MemoryBlockSystem()
        self.utility_manager = UtilityBasedHistoryManager(max_context_tokens)

    async def run(self, user_input):
        # 1. 准备上下文
        context = self.prepare_optimized_context(user_input)

        # 2. LLM 推理
        response = await self.llm.generate(context)

        # 3. 执行工具调用
        while response.has_tool_calls():
            tool_results = await self.execute_tools(response.tool_calls)

            # 4. 智能历史管理
            self.history_manager.add_interaction(
                user_msg=user_input,
                assistant_msg=response,
                tool_calls=response.tool_calls,
                tool_results=tool_results
            )

            # 5. Tool Result Clearing（旧结果）
            optimized_results = self.apply_result_clearing(tool_results)

            # 6. 继续推理
            context = self.prepare_optimized_context(
                user_input,
                previous_response=response,
                tool_results=optimized_results
            )
            response = await self.llm.generate(context)

        # 7. 保存重要信息到 Memory Blocks
        if response.has_key_findings():
            self.memory_blocks.save_to_memory(
                "key_findings",
                response.extract_findings()
            )

        return response.final_answer

    def prepare_optimized_context(self, user_input, **kwargs):
        # 获取分层历史
        recent_history = self.history_manager.get_recent()
        mid_term_summary = self.history_manager.get_mid_term_summary()
        long_term_summary = self.history_manager.get_long_term_summary()

        # 检索相关记忆
        relevant_memories = self.memory_blocks.retrieve_relevant_memory(user_input)

        # 组装上下文
        context = []

        # 长期摘要（如果有）
        if long_term_summary:
            context.append({"role": "system", "content": long_term_summary})

        # 中期摘要
        if mid_term_summary:
            context.append({"role": "system", "content": mid_term_summary})

        # 相关记忆
        if relevant_memories:
            context.append({
                "role": "system",
                "content": f"Relevant memories: {relevant_memories}"
            })

        # 最近完整历史
        context.extend(recent_history)

        # 当前输入
        context.append({"role": "user", "content": user_input})

        # Token 控制
        if self.count_tokens(context) > self.max_context_tokens:
            context = self.utility_manager.prune_to_fit(context)

        return context

    def apply_result_clearing(self, tool_results):
        """清理大体积工具结果"""
        cleared_results = []

        for result in tool_results:
            if len(result["content"]) > 1000:  # 超过阈值
                cleared_results.append({
                    "tool_call_id": result["tool_call_id"],
                    "tool_name": result["tool_name"],
                    "content": f"[Result cleared - {len(result['content'])} chars]",
                    "summary": self.summarize_tool_result(result["content"])
                })
            else:
                cleared_results.append(result)

        return cleared_results

总结

关键要点

  1. 保留 tool 调用信息是必要的

    • 上下文连贯性
    • 防止重复调用
    • 避免 LLM 幻觉

  2. 但需要优化 Token 使用

    • Tool Result Clearing：最直接有效
    • Summarization：平衡性能和上下文
    • Memory Blocks：突破窗口限制
    • Sub-Agent：复杂任务分解

  3. 组合策略最佳

    • 短期：Tool Result Clearing + 滑动窗口
    • 中期：分层历史管理
    • 长期：Memory Blocks + Sub-Agent

实施路径

1
2
3
4
Phase 1 (立即) → Tool Result Clearing
Phase 2 (1-2周) → 滑动窗口 + 摘要
Phase 3 (1个月) → 分层历史管理
Phase 4 (长期) → Memory Blocks + Sub-Agent

参考资源

评分: 4.6/5.0

  • 实用性: 9.5/10 - 所有方案均已产品验证
  • 创新性: 8.5/10 - Tool Result Clearing 是重要创新
  • 可复现性: 9.0/10 - 代码示例完整，易于实现
  • 文档质量: 9.0/10 - 多篇顶会论文支撑
  • 影响力: 9.0/10 - Anthropic、LangChain 等已采用
© 2025 Generative AI Discovery All Rights Reserved.
Theme by hiero