ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Agent上下文引擎完全指南:记忆、压缩与多Agent协作

Agent上下文引擎完全指南:记忆、压缩与多Agent协作 如果你最近在接触 Agent 开发应该会有一个很直观的感受Agent 搭建本身已经没有太多技术门槛了。选一个框架注册几个工具配置好模型半天就能把一个能回答问题的“智能体”跑起来。LangChain、LangGraph、AutoGen、CrewAI 这些框架已经把工具调用、消息解析、多轮对话的组织方式做得相当成熟AI Engineer 真正要拼的已经不是“能不能搭出来”而是“搭出来能不能在真实业务里稳定跑”。真正让 Agent 项目拉开差距的往往是这些问题对话超过十轮之后模型开始忘记前面说过的话工具返回一大段 JSON直接把上下文窗口塞满多个子任务并行执行后主 Agent 无法判断哪个结果对应哪个分支用户换一个会话再进来系统完全不记得他上次的需求。这些问题全部指向同一个核心组件——上下文引擎。这次我们直接对标“AI Engineer”的实战视角把 Agent 搭建和上下文引擎拆开看先确认为什么搭建已经不是难点再给出一套可落地的上下文引擎设计思路然后用代码演示最小 Agent 如何接入记忆、如何做多 Agent 上下文传递、如何服务化支撑批量任务。文中所有工程做法都以通用技术栈为例具体 API 和模型版本按你实际使用的框架调整。1. 为什么说 Agent 搭建已不是难题1.1 Agent 开发已经模板化现在的 Agent 开发本质上是在做“模型能力 工具能力 流程编排”的组合。主流框架已经把下面这些事情封装好了LLM 调用与流式输出Function Calling / Tool Calling 的参数解析多轮对话的 messages 管理简单工具的错误捕获与重试Agent 与外界的代码执行、文件读写、HTTP 请求等能力打通也就是说以前需要自己手写一套“system prompt tools schema 模型返回解析 工具结果再注入”的循环现在框架里有现成的 AgentExecutor、StateGraph、GroupChat 这类抽象开发者只需要定义好工具列表和节点连线。从热搜词也能看到Agent 相关讨论已经从“怎么搭”变成“怎么调”agent框架、agent编排、多agent协作、agent记忆、agent安全。这说明行业共识是——搭建已经模式化工程难点在更深处。1.2 真正的瓶颈上下文引擎搭建解决了“ Agent 能跑”上下文引擎解决的是“Agent 能跑得稳、跑得准、跑得省”。上下文引擎不是一个单一组件而是负责 Agent 运行时上下文信息的获取、组织、存储、检索、压缩、传递与遗忘的系统设计。简单对比一下对比维度Agent 搭建上下文引擎核心问题如何让模型调用工具如何让模型记住关键信息实现手段框架 函数注册存储 检索 压缩 状态管理衡量标准能否跑通一次任务长任务下的一致性与成本失败表现工具调用报错答非所问、上下文污染、token 暴涨调试难度低看日志即可高需要跟踪状态流转很多 Agent 项目在 Demo 阶段表现很好一上生产就“变笨”绝大多数不是模型问题而是上下文没有设计好。上下文引擎做得好同样的模型能发挥出完全不同的效果。2. 上下文引擎核心能力速览能力项说明短期上下文当前会话内的消息队列、工具调用记录、中间推理过程长期记忆跨会话保存用户偏好、历史结论、业务事实工作记忆当前多步骤任务的中间状态、子任务进度、依赖关系上下文压缩对超长历史做摘要、丢弃无效信息、提取关键决策语义检索从历史记忆中召回与当前问题相关的内容避免全量注入上下文隔离不同会话、不同用户、不同子任务之间的数据边界上下文传递主 Agent 与子 Agent、多 Agent 协作时的信息交换规范持久化将上下文写入 Redis / SQLite / PostgreSQL / 向量库支持恢复从实践看一个生产可用的上下文引擎通常要有“读 写 存 压缩 检索”五部分。后面我们逐一落地。3. 上下文引擎设计思路3.1 三类上下文必须分开管理很多 Agent 项目失败是因为把所有信息都塞进 messages 里。正确的做法是把上下文分成三层系统级上下文用户身份、权限、业务规则、固定话术。这部分每次请求都携带但内容相对稳定。会话级上下文当前对话的问答历史、工具返回结果、用户临时给的约束。这部分随会话变化需要做窗口管理和压缩。用户级上下文跨会话沉淀下来的偏好、历史订单、常用参数。这部分需要持久化和检索不能全量塞进提示词。如果三层混在一起短期任务会被历史数据干扰长期记忆又会占用大量 token。3.2 上下文生命周期上下文引擎要覆盖完整生命周期写入 - 存储 - 读取 - 压缩 - 过期清理。一条工具返回结果进来时先判断价值是临时中间结果还是需要长期保存的事实。临时结果只保留在会话级关键事实异步写入用户级记忆。会话结束后对长对话做一次摘要归档关键决策清理过期缓存。这个过程就是 Agent 记忆的“存”和“忘”。4. 环境准备与前置条件以下是一套通用检查清单按你的实际情况调整版本和路径。4.1 基础环境操作系统Linux / macOS / Windows WSL2 均可Python建议 3.10 以上运行时可以访问 OpenAI / Anthropic / 本地推理服务或企业内部模型网关存储Redis用于会话状态、SQLite / PostgreSQL用于结构化记忆、向量数据库用于语义检索可选4.2 安装依赖这里给出一份最小依赖示例实际包名和版本以你使用的框架为准# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # 基础依赖 pip install openai redis fastapi uvicorn # Agent 编排框架按需选择 # pip install langgraph # pip install langchain # pip install autogen-agentchat # 向量检索组件按需选择 # pip install chromadb # pip install qdrant-client4.3 环境变量export OPENAI_API_KEYyour-api-key export REDIS_URLredis://127.0.0.1:6379/0 export AGENT_DATA_DIR./agent_data如果使用本地模型服务把OPENAI_BASE_URL指向本地部署地址export OPENAI_BASE_URLhttp://127.0.0.1:8000/v1从材料看本地部署 Agent 是很多开发者关心的方向。这不影响上下文引擎的设计只影响模型调用层的实现。5. 最小 Agent 带上下文引擎实战5.1 业务目标我们做一个带记忆的客服问答 Agent要求多轮对话中记住用户刚才给过的信息跨会话记住用户偏好工具返回结果不会无限制塞进上下文支持通过 HTTP 接口调用5.2 代码最小 Agent 会话记忆下面代码演示核心思路不依赖特定 Agent 框架。实际项目中可以把这段逻辑嵌入 LangGraph 节点或自定义 Worker。import json import time import redis from openai import OpenAI # 连接 Redis用于会话级和用户级上下文存储 redis_client redis.Redis.from_url(redis://127.0.0.1:6379/0) # 模型客户端按实际模型网关调整 client OpenAI() def get_session_context(session_id: str): 读取当前会话的短期上下文只保留最近 N 条消息 key fsession:ctx:{session_id} raw redis_client.get(key) if not raw: return [] return json.loads(raw) def save_session_context(session_id: str, messages, max_len12): 写入当前会话上下文超过 max_len 时做截断 key fsession:ctx:{session_id} recent messages[-max_len:] redis_client.set(key, json.dumps(recent, ensure_asciiFalse)) redis_client.expire(key, 3600 * 24) def get_user_memory(user_id: str, query: str): 从用户长期记忆中召回相关内容。 简化版直接按关键词匹配生产环境建议用向量检索。 key fuser:memory:{user_id} raw redis_client.get(key) if not raw: return memory json.loads(raw) hits [] for item in memory: if query in item[content] or item[key] in query: hits.append(item[content]) return ; .join(hits[-3:]) def save_user_fact(user_id: str, key: str, content: str): 写入一条用户级记忆使用字典结构便于检索 mem_key fuser:memory:{user_id} raw redis_client.get(mem_key) mem json.loads(raw) if raw else [] # 简单去重同 key 覆盖 mem [m for m in mem if m[key] ! key] mem.append({key: key, content: content, ts: int(time.time())}) # 限制记忆条数防止无限增长 redis_client.set(mem_key, json.dumps(mem[-100:], ensure_asciiFalse)) def run_agent(user_id: str, session_id: str, user_text: str): # 1. 组装系统级上下文 system_prompt 你是一个客服助手回答要简洁、准确。 # 2. 召回用户级记忆 memory_text get_user_memory(user_id, user_text) if memory_text: system_prompt f\n\n用户历史偏好{memory_text} # 3. 读取会话上下文 session_messages get_session_context(session_id) # 4. 调用模型 messages [{role: system, content: system_prompt}] messages.extend(session_messages) messages.append({role: user, content: user_text}) response client.chat.completions.create( modelyour-model-name, # 替换为实际模型 messagesmessages, ) reply response.choices[0].message.content session_messages.append({role: user, content: user_text}) session_messages.append({role: assistant, content: reply}) save_session_context(session_id, session_messages) # 5. 演示当用户主动告知偏好时写入长期记忆 if 我喜欢 in user_text or 偏好 in user_text or 记住 in user_text: save_user_fact(user_id, explicit_preference, user_text) return reply这段代码展示了上下文引擎最基本的四个动作读会话、写会话、召回用户记忆、沉淀用户事实。核心思想是 session 和 user 分开存短期和长期分开管。5.3 运行与验证# 先确保 Redis 已启动 redis-server # 然后运行脚本中的测试逻辑 python agent_demo.py测试输入用户我喜欢简洁的回答请记住这个偏好 助手记录偏好并回复 用户我刚才说了什么偏好如果第二次回答能正确引用“简洁”说明会话上下文和用户记忆都生效了。6. 上下文引擎功能测试6.1 多轮一致性测试测试目的确认 Agent 在长对话中不会丢失关键信息。操作步骤第 1 轮提供业务事实“订单号是 A10086这个订单需要加急”第 5 轮再问“我刚才的订单号是什么需要加急吗”判断回复是否一致失败表现Agent 反问“您什么时候说过订单号”Agent 编造一个不存在的订单号排查方向会话上下文是否做了截断截断时是否把关键事实挤出窗口是否没有把中间结论写入工作记忆6.2 长任务状态恢复测试测试目的确认任务中断后可以恢复。设计思路每个任务分配一个 task_idRedis 中保存任务状态{ task_id: task_001, status: running, current_step: 3, total_steps: 8, intermediate_data: { keyword: 上下文引擎, already_searched: true } }当 Agent 进程重启后读取 task_id 对应的 JSON从第 4 步继续执行。这部分是上下文引擎的“工作记忆”多步骤任务没有它中断一次就得重头再来。6.3 上下文压缩测试测试目的验证超长对话不会撑爆 token。实现思路当会话消息超过阈值时触发摘要用模型将前 N 轮对话压缩成 200 字以内的要点把摘要作为 system 消息的一部分丢弃原始历史def compress_history(messages, max_keep6): 将 messages 中较早的内容压缩为摘要保留最近 max_keep 条 if len(messages) max_keep: return messages old_msgs messages[:-max_keep] recent_msgs messages[-max_keep:] prompt 请将下面的对话历史压缩成要点保留关键事实和用户偏好\n for m in old_msgs: prompt f{m[role]}: {m[content]}\n resp client.chat.completions.create( modelyour-model-name, messages[{role: user, content: prompt}], ) summary resp.choices[0].message.content return [ {role: system, content: f历史摘要{summary}} ] recent_msgs判断成功标准对话超过 30 轮后单位请求 token 数不会持续线性增长同时 Agent 仍能回答早期关键事实。6.4 用户级记忆跨会话测试测试目的确认用户换 session 后仍能复用偏好。操作步骤用户 A 在 session_1 中说“偏好技术深度高的回答”用户 A 在 session_2 中问“我的回答风格偏好是什么”用户 B 在 session_2 中问同样问题应返回无记录判断标准用户隔离正确同一个 user_id 可以跨 session 召回不同 user_id 数据不串。7. 多 Agent 协作中的上下文管理7.1 主从模式与共享上下文多 Agent 协作里最常用的是主从模式主 Agent 负责任务拆解和资源调度子 Agent 负责具体执行。两种上下文设计方式各有场景模式上下文模型适用场景共享上下文所有子 Agent 读取同一份任务上下文团队协作型任务信息需要互相可见隔离上下文每个子 Agent 只拿到自己的输入和输出研发、隐私、权限敏感场景从最近讨论看主从模式本质上可以理解为把 sub-agent 当成一种特殊 tool 来调用。主 Agent 决定何时调用子 Agent子 Agent 接收一段输入返回一段结构化结果然后主 Agent 把这些结果合并进自己的上下文。7.2 子 Agent 上下文传递示意下面用一个函数调用示意多 Agent 之间的上下文传递核心是“输入限定、输出结构化”def research_agent(context: dict, query: str) - dict: 子 Agent 只处理传入的 query 和必要上下文。 返回结构化结果避免把内部过程全部暴露给主 Agent。 # 子 Agent 内部携带自己的短上下文 sub_messages [ {role: system, content: 你是资料检索 Agent只输出结论和来源。}, {role: user, content: query}, ] if context.get(extra_hint): sub_messages.append({role: user, content: context[extra_hint]}) resp client.chat.completions.create( modelyour-model-name, messagessub_messages, ) return { agent: research_agent, result: resp.choices[0].message.content, summary: 已完成资料检索结论见 result 字段, } def main_agent(user_request: str): # 主 Agent 调用子 Agent并把子 Agent 的 summary 注入自己的上下文 sub_result research_agent({}, 请核实Agent 上下文引擎的主流方案有哪些) main_messages [ {role: system, content: 你是主控 Agent综合各子 Agent 结果后给出最终答案。}, {role: user, content: user_request}, {role: system, content: f子Agent已返回结果摘要{sub_result[summary]}}, ] resp client.chat.completions.create( modelyour-model-name, messagesmain_messages, ) return resp.choices[0].message.content这里的关键是子 Agent 的完整内部对话不需要传给主 Agent传一个浓缩后的摘要即可。大量中间过程留在子 Agent 的独立上下文中主 Agent 的上下文才不会被污染。7.3 上下文隔离与权限多 Agent 场景必须考虑权限。子 Agent 的上下文读取范围要遵循最小权限原则调研类 Agent 不应读取订单库财务类 Agent 不应读取用户聊天垃圾箱同一个用户请求并行多个子 Agent 时子 Agent 之间默认不共享内部上下文在上下文引擎里为每个 Agent 设置 context_namespace写入和读取都带命名空间校验。8. 服务化与批量任务8.1 HTTP 接口设计上下文引擎最终要服务化不能只在脚本里跑。下面是一个 FastAPI 风格的服务示例from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): user_id: str session_id: str message: str class ChatResponse(BaseModel): reply: str session_id: str app.post(/v1/chat, response_modelChatResponse) def chat(req: ChatRequest): reply run_agent( user_idreq.user_id, session_idreq.session_id, user_textreq.message, ) return ChatResponse(replyreply, session_idreq.session_id) app.get(/healthz) def healthz(): return {status: ok}启动服务uvicorn api_server:app --host 0.0.0.0 --port 8000测试调用curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d {user_id:u_001,session_id:s_001,message:帮我查一下订单状态}接口层需要注意两点第一session_id 由调用方传入保证同一会话上下文连续第二用户鉴权要在网关完成接口本身不直接信任 user_id。8.2 批量任务队列批量任务场景下上下文引擎不能为每个请求创建新的模型连接应该用任务队列削峰。建议结构输入文件CSV / JSON Lines - 任务队列 - 多个 Worker 并发处理 - 结果写回每个任务需要携带独立的上下文 ID{ task_id: task_0001, user_id: u_001, session_id: batch_001, content: 对待处理内容执行摘要 }Worker 从队列取任务后先读取该 session 的已有上下文再追加本次输入最后写回结果和状态。失败时记录错误并重试避免上下文写入一半造成状态不一致。8.3 并发与状态隔离批量任务最大的坑是上下文串线。多个任务共用同一个 Redis 时必须保证 key 设计带确定性的 ID。推荐 key 设计session:ctx:{session_id} user:memory:{user_id} task:state:{task_id}不要让多个 Worker 使用同一个 session_id 并行写同一份上下文否则会出现覆盖写。每个任务独立 session_id或对写入操作加锁。9. 性能观察与成本控制9.1 token 消耗怎么观察上下文引擎直接影响 token 消耗。建议在每次模型调用前后记录输入 token 和输出 tokenprint(input_tokens:, response.usage.prompt_tokens) print(output_tokens:, response.usage.completion_tokens)观察两个指标单请求输入 token 是否随对话轮数线性增长工具返回结果是否被重复注入如果线性增长很快说明上下文压缩没有生效。9.2 降低延迟和成本的手段压缩历史超出窗口的对话转摘要检索替代全量注入只把与当前问题相关的用户记忆注入工具结果精简调用工具后只取关键字段不把完整 JSON 塞给模型缓存对相同前缀的会话启用结果缓存9.3 资源占用观察使用本地模型时观察资源有一个通用流程# 查看 GPU 显存占用 nvidia-smi # 查看进程状态 top -p $(pgrep -f your_model_worker)显存占用以实际模型版本和推理参数为准。上下文越长KV Cache 越大显存占用随之上升。降低显存最直接的方式是限制上下文长度、减少单批并发数。10. 常见问题与排查方法问题现象可能原因排查方式解决方案多轮对话后 Agent 忘记前面内容会话上下文被截断或未启用检查 Redis 中的 session key 内容增加关键信息摘要关键事实写入用户记忆上下文串线session_id 复用或 key 设计冲突查看日志中写入的 session key每个任务独立 session_idkey 增加命名空间工具返回内容过长token 暴涨工具结果未精简打印工具返回的原始内容长度对工具结果做字段裁剪和摘要用户记忆不区分用户用户级 key 未包含 user_id检查 Redis key 是否存在跨用户覆盖所有用户级 key 必须带 user_id批量任务结果写错多个 Worker 并发写同一 session查看任务日志中的 Worker ID任务独立状态写入加锁接口响应超时模型推理时间长或上下文过短导致多轮补问查看耗时分布增加异步任务接口调用方轮询结果本地模型显存不足上下文窗口设置过长观察 nvidia-smi 输出压缩上下文降低 max_tokens11. 最佳实践与安全边界11.1 工程建议第一次集成上下文引擎时先小参数验证单用户、单会话、50 条消息以内保留一套最小可运行配置出现问题可以做对比上下文存储位置要与代码分离方便迁移和备份批量任务必须加日志、超时和失败重试接口服务要限制访问范围不能默认暴露到公网11.2 安全合规边界上下文引擎存储的是用户对话与业务信息必须注意以下几点用户级记忆写入前要确认授权不能把用户未同意的信息持久化涉及人脸、声音、身份、财务等信息严格按照业务合规要求处理多 Agent 场景下子 Agent 的上下文访问必须有权限控制测试环境使用脱敏数据不要拿真实用户数据跑压测生产环境对上下文存储做加密访问日志定期审计上下文引擎既是效果放大器也是数据风险的集中点。设计阶段不重视权限隔离后期补会非常痛苦。12. 总结与现实建议Agent 搭建已经模板化这确实不是夸大。任何人用主流框架都能在短时间内做出一个能跑的 Agent。但 Demo 和生产的差距大多数时候就是上下文引擎的差距。如果你准备在业务里落地 Agent建议按这个顺序动手先明确系统级、会话级、用户级三层上下文分别存什么用一个 Redis 做会话状态和用户记忆跑通最小链路加多轮一致性测试确认长对话不丢关键事实再考虑多 Agent 协作时的上下文隔离和传递最后做服务化 API 和批量任务最容易踩的坑有三个一是把历史消息无限塞进模型二是多个任务共用 session_id三是用户记忆不做隔离。这三个坑都在上下文引擎的范围内而不在 Agent 搭建本身。上下文引擎应该是 Agent 系统中的一等公民而不是“等出问题了再补”的优化项。把存储、检索、压缩、隔离从一开始设计进去Agent 的上限会高很多后续维护成本也会低很多。建议先从一个会话级上下文 Redis 的最小实现开始跑跑通后再逐步加用户级记忆、语义检索和多 Agent 共享上下文。这条路径比堆框架功能更值得投入。
返回列表