工具内容处理与上下文

Agent 请求模型时, 会把需要参考的消息放进 messages. 对话和工具结果不断累积, 如果始终原样携带, 请求可能超出上下文窗口.

本课参考 demo-02.py 和 NamiAgent 上下文管理讲义, 用三层处理消息: 截断单条工具输出 → 清理旧工具结果 → 压缩旧对话, 再把当前会话保存为 JSON. 三层都是程序执行, 不是模型自行决定; 只有生成摘要需要额外请求模型.

一、先看消息流

一条用户消息标记一轮开始. 一轮可以包含助手回复, 多次 tool_calls 和对应的 tool 结果.

用户输入 → 清理旧工具结果 → 按轮数判断是否摘要压缩 → 请求模型
                                                    ↓
                                        工具调用 → 截断结果 → 回填
                                                    ↓
                                        助手回答 → 保存会话

工具结果在回填前截断; 下一次提问时清理旧结果, 再判断是否压缩对话. 模型获得工具结果后可以继续调用工具, 最终给出回答.

二、第一层: 截断单条工具输出

cut_tool_result(text, limit) 在一条工具结果写入历史之前处理它. 内容没有超过 limit 时原样返回; 超过时保留开头和结尾, 中间插入省略提示. 这样仍能看到文件开头、结尾或错误信息的部分内容, 但中间细节会丢失.

以下是完整函数, 为了突出逻辑, 示例省略了原文件中的 docstring 和行内注释:

def cut_tool_result(text, limit=MAX_CHAR_NUMBER):
    if limit < 0:
        raise ValueError('limit 不能小于 0')
    if len(text) <= limit:
        return text
    start_count = limit // 2
    end_count = limit - start_count
    end = text[-end_count:] if end_count else ''
    return f'{text[:start_count]}\n...中间省略{len(text) - limit}个字...\n{end}'

函数先检查负数, 再处理无需截断的短文本. 对长文本, start_count 和 end_count 分别计算首尾保留数量; 最后一行拼接开头、省略提示和结尾.

例如 limit=4, 0123456789 会保留 01 和 89, 中间提示省略 6 个字符. limit 为奇数时, 末尾多保留一个字符; 为 0 时不保留原文, 负数会报错. 特别注意 text[-0:] 等于完整字符串, 所以代码对 0 单独处理.

limit 限制的是保留的原文字符数, 加上省略提示后结果可能更长. 错误信息也会被截断, 可能导致排错缺少细节.

三、第二层: 清理旧工具结果

clean_old_tool_result(messages, keep) 在下一次用户提问后执行. 它找到所有 role 为 tool 的消息位置, 保留最近 keep 条的内容, 将更早的 content 替换为占位符. 消息本身及 tool_call_id 仍然保留, 不会因为删除工具消息而破坏调用关系.

def clean_old_tool_result(messages, keep=KEEP_RECENT_TOOL_RESULTS):
    tool_msgs = []
    for i, msg in enumerate(messages):
        if msg['role'] == 'tool':
            tool_msgs.append(i)

    if keep > 0:
        old_tool_msgs = tool_msgs[:-keep]
    else:
        old_tool_msgs = tool_msgs

    for i in old_tool_msgs:
        messages[i]['content'] = TOOL_RESULT_PLACEHOLDER

tool_msgs 保存工具消息在原列表中的下标; old_tool_msgs 保存需要处理的下标. 最后的循环根据这些下标修改原列表.

假设有 3 条工具结果, keep=2 时, 第一条内容被替换, 后两条保持原样. keep=0 表示一条也不保留; 需要单独处理, 因为 [:-0] 等同于 [:0], 结果是空列表. 这个函数直接修改传入的 messages, 不返回新列表.

当前示例使用 I am Iron Man 作为占位符, 模型可能误以为它是真实工具输出. 更合适的提示是 [旧工具结果已清理, 如有需要请重新调用工具]. 被替换的原文不能从当前 messages 恢复.

四、第三层: 旧对话摘要压缩

清理工具结果之后, 对话本身仍会持续增长. 摘要压缩分四步: 切分消息 → 整理旧消息 → 请求模型生成摘要 → 组合新历史. 最近几轮原样保留, 较早的消息用摘要代替.

1. 什么是一轮

keep_turns 表示保留的用户轮数. 一条 user 消息标记一轮开始, 后续助手回复和工具结果都属于这一轮, 直到下一条 user 消息. 因此一轮不一定只有一问一答:

user: 请读取文件              ← 一轮开始
assistant: 调用 read_file
tool: 文件内容
assistant: 文件的主要内容是……
user: 再帮我总结              ← 下一轮开始

工具调用和对应结果要一起保留. 如果从它们中间切开, 保留下来的 tool 消息可能找不到对应的调用, 发给接口时会报错. 所以本课选择在 user 消息的位置切分.

2. 找切分点, 划分旧消息和最近消息

classify_old_and_kept_msgs() 先找到所有 user 消息的下标, 再找到倒数第 keep_turns 条用户消息的位置:

def classify_old_and_kept_msgs(messages, keep_turns=DEFAULT_KEEP_TURNS):
    if keep_turns <= 0:
        raise ValueError('keep_turns 必须大于 0')
    user_idx = [i for i, message in enumerate(messages) if message['role'] == 'user']
    if len(user_idx) <= keep_turns:
        return None
    cut = user_idx[-keep_turns]
    return messages[1:cut], messages[cut:]

这里省略了原函数的类型标注和 docstring, 保留完整执行逻辑. keep_turns 必须大于 0; 否则函数主动报错.

注意 user_idx 存的是下标, 最后切分的仍是完整的 messages, 所以助手回复和工具结果不会因为筛选用户消息而消失. 用户消息数量不超过保留轮数时, 没有旧轮次需要压缩, 函数返回 None.

假设 keep_turns=2, 原消息如下:

0  system     系统提示词
1  user       第一轮问题
2  assistant  第一轮回答
3  user       第二轮问题
4  assistant  调用工具
5  tool       工具结果
6  assistant  第二轮回答
7  user       第三轮问题

user_idx 为 [1, 3, 7], user_idx[-2] 得到 3, 因此 cut=3:

• messages[1:3]: 下标 1、2, 即需要压缩的第一轮. 从 1 开始是为了跳过 system 消息.
• messages[3:]: 从第二轮开始直到末尾, 即保留的最近两轮, 其中包括工具调用和结果.

当程序在发起模型请求前处理刚输入的新问题时, 最后一轮可能只有 user 消息, 尚未有助手回复. 这仍然是最近一轮.

3. compact 的完整流程

compact() 把拆分、整理、摘要和重新组合连起来. 下面保留完整执行逻辑, 省略 docstring 和行内注释:

def compact(messages):
    parts = classify_old_and_kept_msgs(messages=messages)
    if parts is None:
        return messages
    old, kept = parts

    summaried_old = render_for_summary(old)
    response = client.chat.completions.create(
        model=MODEL,
        messages=[
            {'role': 'system', 'content': SUMMARY_PROMPT},
            {'role': 'user', 'content': summaried_old},
        ],
    )
    summary = response.choices[0].message.content
    if not summary or not summary.strip():
        print('摘要为空, 保留原对话.')
        return messages
    print(f'\n[上下文压缩] 已将 {len(old)} 条旧消息整理为摘要.\n摘要: {summary}')

    return [
        messages[0],
        {'role': 'user', 'content': f'[以下是此前对话的摘要]\n{summary}'},
        {'role': 'assistant', 'content': '好的，我已了解之前的进展，继续。'},
        *kept,
    ]

下面按执行顺序看每一步. 首先接收拆分结果:

parts = classify_old_and_kept_msgs(messages=messages)
if parts is None:
    return messages

old, kept = parts

parts 不为 None 时是一个元组: 第一个元素是旧消息列表 old, 第二个元素是保留的消息列表 kept. 这一步只做拆分, 还没有请求模型. 如果无需压缩, 函数直接返回原消息, 不产生摘要请求.

4. 整理旧消息并请求模型总结

旧消息是一个由字典组成的列表. render_for_summary(old) 将每条消息的角色和文字内容整理成文本; 对带 tool_calls 的助手消息, 还会写出工具名和参数. 完整执行逻辑如下:

def render_for_summary(msgs) -> str:
    lines = []
    for m in msgs:
        role = m['role']
        content = m.get('content', '')
        if content:
            lines.append(f'[{role}] {content}')
        if role == 'assistant' and m.get('tool_calls'):
            for call in m.get('tool_calls'):
                name = call['function']['name']
                args = call['function']['arguments']
                lines.append(f'[assistant 调用工具] {name}({args})')
    return '\n'.join(lines)

循环逐条处理消息, 有内容就添加带角色的一行; 助手有工具调用时再逐个记录工具名和参数. 最后 "\n".join(lines) 用换行符把各行连接成一个字符串, 例如:

[user] 帮我读取文件
[assistant 调用工具] read_file({"path": "a.txt"})
[tool] 文件内容

之后用这段文本单独请求模型生成摘要:

summaried_old = render_for_summary(old)
response = client.chat.completions.create(
    model=MODEL,
    messages=[
        {'role': 'system', 'content': SUMMARY_PROMPT},
        {'role': 'user', 'content': summaried_old},
    ],
)
summary = response.choices[0].message.content

这次请求中的 system 消息说明摘要要求, user 消息提供待总结的旧对话. summary 是模型返回的摘要文本. 这与后面请求模型回答用户是两次请求; 生成摘要会增加一次模型调用.

如果模型返回的摘要为空, 当前实现会返回原始 messages, 避免用空摘要替换旧对话:

if not summary or not summary.strip():
    print('摘要为空, 保留原对话.')
    return messages

5. 重新组合消息列表

摘要生成成功后, 将原 system 消息、摘要和最近几轮拼起来:

return [
    messages[0],
    {'role': 'user', 'content': f'[以下是此前对话的摘要]\n{summary}'},
    {'role': 'assistant', 'content': '好的，我已了解之前的进展，继续。'},
    *kept,
]

messages[0] 保留原系统提示词; 摘要放在一条新的 user 消息中; 固定的 assistant 回复由代码写入, 不是模型生成的; *kept 把最近几轮的消息逐条展开. 外层通过 history = compact(history) 接收新列表, 后续请求就使用这份上下文.

6. 串起来看一次压缩

假设用户依次说了三句话: 我叫小林, 我的项目叫 LyrjAgent, 我正在学习上下文压缩. 设置 keep_turns=2 后, 第一轮进入 old, 后两轮进入 kept. 假设第一轮助手回复了 你好, 小林, 则整理出的旧消息文本是:

[user] 我叫小林
[assistant] 你好, 小林

假设摘要模型返回 用户叫小林, 新历史依次包含原 system 消息、一条内容为 [以下是此前对话的摘要]\n用户叫小林 的 user 消息、一条固定 assistant 回复, 以及 kept 里的后两轮原文. 因此后续模型看到的是摘要和最近对话, 不再是第一轮的完整原文. 这个摘要是示意, 实际结果由模型生成.

摘要可能遗漏或误解信息, 再次压缩时旧摘要也可能被重新总结. 摘要以 user 消息加入历史, 后续计数时也会被当作一轮.

本课示例按用户消息数量触发摘要, 超过 DEFAULT_KEEP_TURNS 才额外请求模型. 参考讲义介绍的是根据 API 返回的 usage.prompt_tokens 设置阈值; 当前练习没有实现 token 阈值判断.

五、会话持久化: 保存和恢复

程序运行时的 messages 是列表, 列表中的消息是字典. save_session() 把当前历史序列化为 UTF-8 JSON, load_session() 在程序启动时读回; 找不到文件则从一条 system 消息开始新会话.

def save_session(history, path=SESSION_FILE):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(history, ensure_ascii=False, indent=2), encoding='utf-8')


def load_session(path=SESSION_FILE):
    if not path.exists():
        return None
    history = json.loads(path.read_text(encoding='utf-8'))
    if not isinstance(history, list) or not history or not isinstance(history[0], dict) or history[0].get('role') != 'system':
        raise ValueError('会话文件必须包含以 system 消息开头的消息列表')
    return history

save_session() 用 mkdir 创建父目录, 再将历史写入文件; ensure_ascii=False 让中文直接写入, indent=2 方便阅读. load_session() 找不到文件时返回 None, 否则读取并解析 JSON. 它还检查结果是非空列表、第一项是字典且角色为 system; 这只是基本检查, 不会逐条校验其他消息.

文件默认位于 ~/.lyrj-agent/sessions/default.json, 即用户目录下的程序状态目录, 而不是依赖启动时所在的目录. 每轮成功回复后覆盖保存, 下次运行继续读取; 模型请求或写文件失败时, 当前轮未必已经保存.

文件保存的是当时的上下文, 不是完整原始聊天档案: 被替换的旧工具内容和被摘要替代的对话不会自动恢复. JSON 文件可能包含对话内容和工具参数, 不应随意提交或分享. 当前练习没有为文件工具提供工作区沙箱, 因此把会话文件放在项目目录之外, 不代表 Agent 一定无法访问它.

六、怎样观察效果

测试时可将三个参数分别设为 80, 2, 3, 保存后重启. 读取长文件观察省略提示; 多次调用工具后, 再提问并在 JSON 中查找占位符; 轮数超限时查看终端摘要. 单独测试第二层时可先调大保留轮数.