AI Agent 工具调用大结果处理(5 种实战方案对比)

工具调用返回超大结果,是 Agent 工程里最常被低估的”长尾问题”——一个数据库查询工具可能返回几万行,一段代码执行可能输出几十 MB,一个文档抓取可能拿到一整本电子书。如果让 LLM 直接消费这些原始数据,轻则 context 爆掉、重则模型输出失控。在做日志分析 Agent 时,我曾让 Agent 直接”读完整日志文件”来定位异常,结果一条 800 MB 的日志把上下文撑爆,模型开始胡言乱语;改用”采样 + 摘要 + 关键段提取”三层处理后,同样的任务跑得又稳又快。下文把 5 种实战方案放在一起对比,给出适用场景和工程权衡。

一、问题为什么严重

Agent 的工作循环是”调工具 → 拿结果 → 让 LLM 推理 → 决定下一步”。工具返回的原始数据通常远大于 LLM 上下文窗口:

数据库查询:一条 SELECT * FROM events WHERE date > '2024-01-01' 可能返回几万行;
代码执行:python script.py 可能输出几 MB 的 stdout;
API 调用:GET /api/v1/users 一次性返回全量用户列表;
文档读取:read_file 一份 100 页 PDF 一次性返回全文;
网页抓取:fetch_url 拿到整个 HTML 树。

把这些原始数据直接塞进 LLM 的 prompt,会立刻触发三个连锁问题:上下文超限 → 模型截断/报错 → 决策质量崩塌。

二、5 种处理方案对比

下面是工程上常见的 5 种方案,各自的适用场景和代价:

方案	核心思路	适用场景	优点	缺点
截断	取前 N 字符/行	结果前部最有价值	极简、零成本	信息丢失严重
采样	随机/分层取样	大数据集有代表性	保留分布特征	仍可能漏掉关键点
摘要	LLM 二次压缩	文本类结果	保留语义	额外 token 成本、有损
分块	拆成多段,逐步消费	文档/长文本	全量覆盖	增加循环次数
落盘 + 检索	结果存外部,按需检索	反复使用的大结果	单次便宜、可复用	增加架构复杂度

三、方案 1:截断(Truncation)

最简单粗暴:取前 N 字符或前 N 行,丢弃其余。在 Agent 框架里,通常作为兜底方案自动启用。

def truncate_result(content: str, max_chars: int = 4000) -> str:
    if len(content) <= max_chars:
        return content
    return content[:max_chars] + f"\n\n[已截断,共 {len(content)} 字符,仅显示前 {max_chars}]"

适用场景:结果前部确实最有价值(比如错误日志的开头、API 返回的元信息)。不适用:关键信息可能藏在中间或尾部(比如会议纪要的核心决策在最后)。

四、方案 2:采样(Sampling)

对大数据集,按规则取代表性样本:随机采样、分层采样、首尾采样、关键字段过滤。

import random

def stratified_sample(rows, n=100, key_fn=None):
    if len(rows) <= n:
        return rows
    if key_fn is None:
        return random.sample(rows, n)
    # 按 key 分层,每层取前 k 个
    buckets = {}
    for r in rows:
        k = key_fn(r)
        buckets.setdefault(k, []).append(r)
    per_bucket = max(1, n // len(buckets))
    out = []
    for v in buckets.values():
        out.extend(v[:per_bucket])
    return out[:n]

适用场景:统计查询、列表浏览、批量数据探索。优点:保留整体分布。缺点:仍可能漏掉异常值或边缘 case,任务关键时不能依赖。

五、方案 3:摘要(Summarization)

用 LLM 二次压缩结果,保留语义信息。工程上要注意两点:分段摘要(避免一次摘要超限)+ 关键字段保留(强制把 ID、时间、状态码等”硬信息”留在摘要里)。

def summarize_result(client, content, max_tokens=500):
    # 先粗切,避免单次 prompt 超限
    chunks = [content[i:i+8000] for i in range(0, len(content), 8000)]
    summaries = []
    for ch in chunks:
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "把这段内容压缩到 200 字以内,保留所有 ID、错误码、时间戳等硬信息。"},
                {"role": "user", "content": ch}
            ]
        )
        summaries.append(resp.choices[0].message.content)
    # 二次合并
    return "\n".join(summaries)

适用场景:文本类结果(日志、文档、API JSON 文本)。代价:多花一份 token 成本,且摘要本身有损(可能漏掉对当前任务关键的细节)。

六、方案 4:分块 + Map-Reduce(Chunking + Map-Reduce)

把大结果拆成多块,每块单独让 LLM 处理,最后合并结论。这是处理”必须覆盖全量数据”场景的稳妥路径。

def map_reduce_result(client, content, question, chunk_size=6000):
    chunks = [content[i:i+chunk_size] for i in range(0, len(content), chunk_size)]
    # Map: 每块独立回答
    partial_answers = []
    for i, ch in enumerate(chunks):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": f"你是分析师。回答: {question}\n如果有相关信息就回答,没有就说'本块无相关信息'。"},
                {"role": "user", "content": ch}
            ]
        )
        partial_answers.append(f"块{i+1}: {resp.choices[0].message.content}")
    # Reduce: 综合
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"综合多块分析员结论,回答原问题: {question}"},
            {"role": "user", "content": "\n\n".join(partial_answers)}
        ]
    ).choices[0].message.content

适用场景:需要从大文档/大日志里精确找答案(比如”找出所有 4xx 错误的具体 URL”)。代价:循环次数多、token 成本高,慢但稳。

七、方案 5:落盘 + 按需检索(Persist + Retrieve)

把工具结果存到外部存储(向量库/文件系统/数据库),Agent 后续只检索需要的片段。这是最贴近”工具 + 记忆”设计哲学的方案。

import chromadb

client_db = chromadb.PersistentClient(path="./agent_memory")
collection = client_db.get_or_create_collection("tool_results")

def persist_result(tool_name, content, metadata):
    import uuid
    doc_id = f"{tool_name}_{uuid.uuid4()}"
    collection.add(documents=[content], ids=[doc_id], metadatas=[metadata])
    return doc_id

def retrieve_relevant(query, top_k=5):
    results = collection.query(query_texts=[query], n_results=top_k)
    return "\n\n".join(results["documents"][0])

Agent 拿到工具结果后,先持久化(返回 doc_id),再决定下一步是否需要进一步检索。适用场景:同一个大结果会被 Agent 在多步任务里反复使用,或跨会话复用。

八、方案选型决策树

按下面三步选,基本不会错:

结果大小: < 8K 字符 → 直接用;8K-80K → 截断或摘要;> 80K → 落盘 + 检索。
任务性质: 浏览性查询 → 采样;精确查找 → 分块 Map-Reduce;反复使用 → 落盘 + 检索。
成本预算: 极敏感 → 截断/采样;宽松 → 摘要/分块;不在乎 → 落盘 + 检索。

实战里多数项目用”截断作兜底 + 摘要作默认 + 落盘 + 检索作进阶”三段式:小结果直接消费、中结果摘要、大结果落盘按需取。

常见问题(FAQ)

Q1：为什么不让 LLM 自己决定怎么处理大结果?

LLM 在长上下文里”判断”自己已经超限的准确率不高,通常会继续接收直到报错;由工程层强制处理更稳。

Q2：Map-Reduce 的成本大概是多少?

通常是大结果直接喂 LLM 的 1.5-3 倍,但准确率提升明显,关键任务值得。

Q3：落盘 + 检索的延迟来自哪?

向量检索本身(<100ms)+ embedding 计算(如果现算)+ Agent 多步决策;通常比 Map-Reduce 快,但架构最复杂。