ARTICLE DETAIL

资讯详情

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

Agent Harness 实战:上下文管理与编排,让智能体稳定运行

Agent Harness 实战:上下文管理与编排,让智能体稳定运行 1. 从能跑到跑得住Agent Harness 到底在解决什么很多人第一次接触智能体开发都是从一段几十行的 ReAct 循环开始的把工具描述塞进系统提示词让模型输出 Thought / Action / Observation解析文本、调用工具、把结果拼回上下文循环到模型给出 Final Answer。这段代码在 Demo 阶段几乎百试百灵但只要任务稍微长一点、工具稍微多一点问题就会集中爆发——模型开始忘记前面做过什么、重复调用同一个工具、把工具返回的 JSON 当成用户说的话、上下文窗口被无关内容撑爆、一次失败之后整个循环卡死。这些问题的根源不在模型本身而在于缺少一层专门负责驾驭的运行时。这层运行时就是现在被反复提到的Agent Harness。直译过来是智能体挽具你可以把它理解成马具模型是那匹有力气但方向感不稳定的马Harness 是缰绳、鞍具和车架的组合负责把模型的推理能力约束在一条可控的轨道上同时把外部世界的反馈稳定地喂回去。需要先厘清一个高频混淆点Harness 和 Agent 不是一回事。Agent 是能思考、能行动的那个决策主体它的核心是模型加提示词加工具集Harness 是承载这个主体的运行时框架它管的是上下文怎么组装、消息怎么流转、工具怎么调度、状态怎么持久化、错误怎么恢复、循环怎么终止。用一句话概括Agent 决定做什么Harness 决定怎么稳定地做下去。市面上很多号称Agent 框架的东西其实一大半代码都在做 Harness 的活。这篇笔记围绕上下文管理与编排这两条主线展开把我在实际搭建 Agent Harness 过程中踩过的坑、验证过的结构、以及那些文档里不会写的细节整理出来。内容适合已经写过基础 ReAct 循环、想让自己的智能体从玩具变成能连续跑几十轮不崩的开发者也适合正在选型编排方案、纠结要不要上重型框架的人。全程以 OpenAI SDK 风格的消息协议为基准因为它的messages数组结构最贴近底层理解它之后再看任何上层框架都不会迷路。2. 上下文管理Harness 里最容易失控的那部分2.1 消息数组不是日志是有结构的运行时状态新手最容易犯的错是把messages当成一个只增不减的日志数组每轮往后面 append 就完事。跑十几轮之后你会发现模型的行为开始变得诡异明明上一轮已经确认过的参数这一轮又问一遍工具返回的错误信息被它当成了新的用户指令。原因在于消息数组是有角色语义的结构化状态不是流水账。OpenAI SDK 风格的消息角色大致分四类system承载全局指令和工具定义user承载人类输入assistant承载模型输出包括它发起的工具调用tool承载工具执行结果。这四类消息在模型眼里权重完全不同。当你把工具返回的一大坨原始数据直接塞进tool角色模型会认真对待它但如果你图省事把工具结果拼成一段字符串塞进user角色模型就会把它当成用户又说了句话语义直接错位。我在 Harness 里做的第一件事就是给消息数组加一层类型化的封装而不是裸操作字典。每条消息进数组之前都要过一遍校验角色是否合法、tool_call_id是否和前面的assistant消息对得上、内容是否为空。这一步看起来啰嗦但它拦住了后面 80% 的诡异行为。def append_message(history, role, content, tool_call_idNone, nameNone): if role not in (system, user, assistant, tool): raise ValueError(f非法角色: {role}) if role tool and not tool_call_id: raise ValueError(tool 消息必须携带 tool_call_id) msg {role: role, content: content} if tool_call_id: msg[tool_call_id] tool_call_id if name: msg[name] name history.append(msg) return history提示tool_call_id的配对关系是硬约束。一旦出现一个tool消息找不到对应的assistant.tool_calls不同模型供应商的反应不一样有的直接报错有的默默忽略有的会开始胡言乱语。宁可在这里抛异常也不要让它悄悄溜过去。2.2 上下文窗口的预算分配给谁留位置上下文窗口是有限资源Harness 必须像管内存一样管它。我的做法是给窗口划出几个固定预算区而不是等快满了再临时裁剪预算区建议占比内容是否可裁剪系统指令区10%~15%角色设定、行为约束、输出格式否工具定义区15%~25%工具 schema、参数说明否近期对话区40%~50%最近 N 轮完整消息是长期记忆区10%~20%摘要、关键事实是输出预留区10%~15%留给模型生成本轮回复否这个分配不是拍脑袋来的。系统指令和工具定义是 Harness 的宪法一旦被裁掉模型的行为约束就没了所以它们永远不参与裁剪。输出预留区必须留够否则模型生成到一半被截断会返回一个残缺的tool_calls解析直接失败。真正可以动的只有近期对话区和长期记忆区。我踩过的一个坑是早期为了塞进更多历史把工具定义压缩成一句话描述。结果模型开始乱传参数因为它根本不知道每个参数的类型和取值范围。工具 schema 的完整性优先级高于历史长度这个顺序不能反。2.3 裁剪策略滑动窗口、摘要、还是混合裁剪历史有几种常见策略各有适用场景滑动窗口只保留最近 N 轮实现最简单但会丢失早期关键信息。适合短任务、无状态场景。摘要压缩把早期对话交给模型总结成一段话替换掉原始消息。省 token但摘要本身有信息损失且多一次模型调用。关键事实抽取从历史里抽出结构化的事实比如用户的城市是杭州订单号是 A123单独存成键值对需要时再注入。信息密度最高但抽取逻辑要自己写。混合策略近期用滑动窗口保完整远期用摘要或事实抽取保要点。我最终采用的是混合策略触发条件是当消息数组的估算 token 数超过窗口的 70%。为什么是 70% 而不是 90%因为工具返回的结果长度不可预测一次搜索可能返回几千 token留 30% 的缓冲能避免裁剪完立刻又超的抖动。def estimate_tokens(messages): # 粗略估算中文约 1.5 字/token英文约 4 字符/token total 0 for m in messages: content m.get(content) or total len(content) // 2 10 # 10 为角色等固定开销 return total def maybe_compress(history, window_limit, keep_recent6): if estimate_tokens(history) window_limit * 0.7: return history system_msgs [m for m in history if m[role] system] dialog [m for m in history if m[role] ! system] recent dialog[-keep_recent:] old dialog[:-keep_recent] if old: summary summarize(old) # 调用模型生成摘要 summary_msg {role: system, content: f[历史摘要] {summary}} return system_msgs [summary_msg] recent return history注意摘要消息我习惯用system角色而不是user因为它是 Harness 注入的元信息不是人类说的话。用user角色会让模型误以为用户在复述历史容易引发混乱。2.4 工具结果的瘦身别把原始数据全喂回去工具返回的结果往往是上下文膨胀的最大元凶。一次网页抓取可能返回几万字符的 HTML一次数据库查询可能返回上百行记录。如果原样塞回上下文几轮下来窗口就爆了。我的处理原则是工具结果在进入上下文之前必须先经过一层 Harness 侧的加工。加工包括三步截断、结构化、标注来源。截断不是简单砍掉尾部而是保留头部和关键字段。比如网页抓取我会提取正文文本、去掉标签、限制在 2000 字符以内并在末尾标注[内容已截断原始长度 15234 字符]。这个标注很重要它让模型知道信息不完整避免它基于残缺信息下结论。结构化是把工具返回的 JSON 转成模型更容易理解的紧凑格式。比如一个查询返回 50 条记录我不会全塞进去而是先做聚合告诉模型共 50 条以下是前 5 条样例和字段统计。模型需要的是知道有什么而不是看到全部。def shrink_tool_result(raw, max_chars2000): text extract_text(raw) # 从 HTML/JSON 中提取可读文本 if len(text) max_chars: return text head text[:max_chars] return f{head}\n[内容已截断原始长度 {len(text)} 字符]这套瘦身逻辑我放在 Harness 的工具调度层而不是让每个工具自己实现。原因是工具作者往往只关心返回正确数据不关心上下文预算把瘦身统一收口到 Harness 才能保证一致性。3. 编排让 ReAct 循环从能转到转得稳3.1 ReAct 循环的骨架与它的三个脆弱点ReAct 的核心就三步Reason推理→ Act行动→ Observe观察循环往复直到得出答案。骨架简单到可以手写但真正跑起来脆弱点集中在三个地方。第一个脆弱点是解析。模型输出的tool_calls结构在大多数情况下是规范的但偶尔会返回格式错误的 JSON、或者把多个工具调用塞进一个不规范的字符串里。如果 Harness 直接json.loads而不做容错一次解析失败就会让整个循环崩掉。第二个脆弱点是终止条件。ReAct 靠模型自己决定什么时候输出 Final Answer。但模型有时候会陷入我再查一下的循环或者过早地给出一个不完整的答案。Harness 必须设置硬性上限最大轮数、最大工具调用次数、最大 token 消耗任何一个触顶就强制收尾。第三个脆弱点是错误传播。工具调用失败时如果直接把异常抛出去循环就断了如果把错误信息原样塞回上下文模型可能反复重试同一个失败调用。正确的做法是把错误结构化后返回并附带重试建议。def run_react_loop(agent, tools, max_turns15, max_tool_calls30): turns 0 tool_calls 0 while turns max_turns: turns 1 response agent.chat() if response.finish_reason stop: return response.content for call in response.tool_calls: tool_calls 1 if tool_calls max_tool_calls: return 已达到工具调用上限基于现有信息给出结论。 result safe_execute(tools, call) agent.append_tool_result(call.id, result) return 已达到最大轮数基于现有信息给出结论。3.2 工具调度的幂等与超时那些让循环卡死的细节工具调度看起来就是拿到名字、找到函数、执行、返回结果但实际生产里卡死循环的往往是这些细节。超时是必须的。一个没有超时的 HTTP 请求遇到对端不响应会一直挂着整个 Agent 循环就停在那里。我给每个工具调用都套了超时默认 30 秒可配置。超时后返回一个结构化的错误让模型知道这个工具这次没成功而不是让它无限等待。幂等是容易被忽略的。模型在重试时可能会重复调用一个有副作用的工具比如发送邮件创建订单。如果工具本身不幂等就会产生重复操作。我的做法是在 Harness 层给每个工具调用生成一个基于参数哈希的call_key在同一个会话内相同call_key的调用直接返回缓存结果不再真正执行。import hashlib, json def call_key(tool_name, args): raw json.dumps({tool: tool_name, args: args}, sort_keysTrue) return hashlib.md5(raw.encode()).hexdigest() def safe_execute(tools, call, cacheNone, timeout30): key call_key(call.name, call.arguments) if cache is not None and key in cache: return cache[key] try: result run_with_timeout(tools[call.name], call.arguments, timeout) if cache is not None: cache[key] result return result except TimeoutError: return {error: timeout, hint: 该工具超时可尝试换用其他工具或简化参数} except Exception as e: return {error: str(e), hint: 调用失败请检查参数或换用其他方式}提示缓存只对读类工具开启写类工具发送、创建、删除不要缓存否则会掩盖真实的副作用。判断标准是工具是否有外部可见的状态变更。3.3 并行工具调用省时间但别省掉顺序依赖现代模型支持一次返回多个tool_callsHarness 可以并行执行它们来省时间。但并行有个前提这些调用之间没有顺序依赖。如果模型同时调用了创建订单和查询订单状态而后者依赖前者的结果并行就会出错。我的处理方式是默认并行执行同一轮内的所有工具调用但在工具定义里加一个depends_on字段声明依赖关系。Harness 在执行前先做一次拓扑排序有依赖的串行无依赖的并行。def execute_batch(tools, calls): # 简化版按 depends_on 分组无依赖的并行 independent [c for c in calls if not tools[c.name].get(depends_on)] dependent [c for c in calls if tools[c.name].get(depends_on)] results {} with ThreadPoolExecutor() as pool: futures {pool.submit(safe_execute, tools, c): c for c in independent} for fut in as_completed(futures): c futures[fut] results[c.id] fut.result() for c in dependent: results[c.id] safe_execute(tools, c) return results实测下来并行执行对多源信息检索这类场景提速明显一次查三个数据源耗时从 3 倍降到 1 倍多一点。但对有副作用的写操作我倾向于保守宁可串行也不冒顺序错乱的风险。3.4 状态持久化让 Agent 能断点续跑长任务跑到一半进程重启了怎么办如果状态只在内存里一切从头再来。Harness 需要把关键状态持久化至少包括消息历史、已执行的工具调用记录、当前轮数、缓存。我用的方案很朴素每轮循环结束后把状态序列化成 JSON 落盘key 用会话 ID。重启时先尝试加载加载成功就从断点继续。这里有个细节工具调用的副作用不能重放。所以持久化时要记录哪些工具已经真正执行过恢复时对这些调用直接返回记录的结果而不是重新执行。def save_state(session_id, state): with open(fstate/{session_id}.json, w) as f: json.dump(state, f, ensure_asciiFalse) def load_state(session_id): path fstate/{session_id}.json if os.path.exists(path): with open(path) as f: return json.load(f) return None这套机制让 Agent 具备了断点续跑的能力对于动辄跑几分钟甚至更久的长任务价值很大。4. 把 Harness 和编排串起来一个可复用的最小实现4.1 分层结构别把编排逻辑和业务逻辑搅在一起我见过太多项目把 ReAct 循环、工具实现、业务规则全写在一个文件里改一处牵动全身。正确的做法是分层协议层定义消息结构、工具 schema、状态格式。这一层不依赖任何具体模型。Harness 层上下文管理、裁剪、工具调度、循环控制、持久化。这一层不依赖具体业务。Agent 层系统提示词、工具集、业务规则。这一层是每个项目独有的。应用层对外接口、会话管理、用户交互。分层的价值在于Harness 层可以跨项目复用。我现在的做法是把 Harness 抽成一个独立模块新项目只需要写 Agent 层的提示词和工具Harness 直接拿来用。省下来的时间非常可观。4.2 一个最小可用的 Harness 骨架下面这个骨架去掉了业务细节保留了 Harness 的核心结构可以直接作为起点class AgentHarness: def __init__(self, agent, tools, window_limit8000, max_turns15): self.agent agent self.tools tools self.window_limit window_limit self.max_turns max_turns self.history [] self.cache {} self.executed {} def run(self, user_input): append_message(self.history, user, user_input) for turn in range(self.max_turns): self.history maybe_compress(self.history, self.window_limit) response self.agent.chat(self.history, self.tools) append_message(self.history, assistant, response.content, tool_callsresponse.tool_calls) if not response.tool_calls: return response.content for call in response.tool_calls: if call.id in self.executed: result self.executed[call.id] else: result safe_execute(self.tools, call, self.cache) self.executed[call.id] result append_message(self.history, tool, shrink_tool_result(result), tool_call_idcall.id) return 达到最大轮数基于现有信息收尾。这个骨架不到 30 行但已经包含了上下文压缩、工具调度、缓存、断点记录、轮数上限这几个关键能力。你可以在此基础上按需扩展。4.3 编排模式的选择ReAct 不是唯一答案ReAct 适合边想边做、路径不确定的任务比如开放式研究、多步排查。但并不是所有任务都适合 ReAct。如果你的任务流程是固定的比如先查用户、再查订单、再生成报告用workflow 编排把步骤写死每步调一次模型反而更稳、更省 token、更容易调试。我的经验判断标准是任务特征推荐模式步骤固定、依赖明确Workflow 编排路径不确定、需要探索ReAct需要多轮反思、自我纠错ReAct 反思步骤多 Agent 协作编排层调度多个 Harness 实例很多团队一上来就上 ReAct结果发现任务其实很简单模型绕了一大圈还容易出错。先用 workflow 把能固定的固定下来只在真正需要探索的环节用 ReAct这是我踩过坑之后最想分享的一条经验。5. 那些文档里不会写的实操心得5.1 提示词里的工具描述比工具实现更影响成功率我做过一个对比实验同一个工具实现完全不变只改提示词里的描述成功率能差出 30%。工具描述要写清楚三件事什么时候用、参数怎么填、返回什么。尤其是什么时候用很多工具描述只写了功能没写触发条件模型就不知道该不该调。举个例子一个搜索工具如果描述只写搜索信息模型可能在任何时候都想搜一下。如果写成当需要获取实时信息或你不确定的事实时使用不要用于你已经知道答案的问题模型的调用就精准多了。5.2 错误信息要写给模型看不是写给人看工具报错时str(exception)往往是一堆堆栈模型看不懂。我会把错误包装成模型能理解的形式{error: 参数缺失, missing: city, hint: 请补充城市名称后重试}。这样模型下一轮就知道该补什么而不是对着堆栈发呆。5.3 循环终止不能只靠模型自觉模型有时候会上瘾一直觉得信息不够反复调用工具。硬性上限是必须的但上限触发时的收尾方式也有讲究。直接抛异常会让用户看到错误更好的做法是注入一条系统消息已达到工具调用上限请基于现有信息给出最佳答案让模型优雅收尾。5.4 日志要记全但别记进上下文调试 Agent 最痛苦的是不知道它为什么这么决策。我的做法是把每一轮的完整消息、模型原始输出、工具调用参数和结果都记到独立日志文件但这些日志不进上下文。上下文只保留精简后的版本。日志用于事后复盘上下文用于模型决策两者分开。5.5 上下文压缩的时机比策略更重要前面提到 70% 触发压缩这个阈值我调过很多次。太低比如 50%会导致频繁压缩摘要调用本身消耗 token 和时间太高比如 90%会导致压缩后立刻又超抖动明显。70% 是我在多个项目里验证下来比较稳的平衡点但具体项目还要根据工具返回的平均长度微调。6. 从单 Agent 到多 AgentHarness 的扩展边界当任务复杂到单个 Agent 搞不定时自然会想到多 Agent 协作。但多 Agent 不是简单地把几个 Harness 实例拼在一起它引入了新的问题Agent 之间怎么通信、状态怎么共享、冲突怎么仲裁。我的做法是给每个 Agent 配一个独立的 Harness 实例各自管理自己的上下文Agent 之间通过一个编排层通信。编排层负责路由消息、汇总结果、处理冲突。这样每个 Agent 的上下文是隔离的不会互相污染编排层则专注于协调。class Orchestrator: def __init__(self, agents): self.agents agents # {name: AgentHarness} def dispatch(self, task): # 简单路由按任务类型分派 if research in task: return self.agents[researcher].run(task) if write in task: context self.agents[researcher].last_result return self.agents[writer].run(f{task}\n参考资料{context}) return self.agents[general].run(task)多 Agent 的坑在于上下文传递。研究者 Agent 的完整上下文可能有几千 token全传给写作者 Agent 会撑爆后者的窗口。我的做法是只传结论和关键事实不传过程。这又回到了前面说的关键事实抽取在多 Agent 场景下这个能力的价值被放大了。7. 我个人的几条经验总结搭 Agent Harness 这件事技术难度其实不高难的是对边界的判断哪些交给模型哪些必须由 Harness 兜底。我的原则是凡是涉及正确性、安全性、资源上限的一律由 Harness 硬性控制不依赖模型自觉凡是涉及内容生成、路径选择、信息理解的交给模型发挥。上下文管理的核心不是塞得多而是塞得准。一条精准的系统消息价值可能超过十轮完整对话。工具调度的核心不是调得快而是调得稳超时、幂等、错误结构化这三件事做到位循环的稳定性会有质的提升。最后说一个心态上的体会不要追求一步到位。我最初的 Harness 只有几十行随着踩坑一点点加功能现在也不过几百行。每次加功能都是因为遇到了具体问题而不是因为框架应该有这个。这种由问题驱动的演进方式比一开始就设计一个大而全的框架要靠谱得多。你完全可以从第 4 节那个最小骨架开始跑起来遇到问题再补这样长出来的 Harness 才是真正贴合你需求的。
返回列表