首页 / 资讯中心 / 文章详情

从单次LLM调用到Harness Runtime:Agent架构选型与工程实践

从单次LLM调用到Harness Runtime:Agent架构选型与工程实践 ★ FEATURED ARTICLE
1. 从一次裸调用说起为什么单次 LLM 调用撑不起一个 Agent很多人第一次接触 Agent 开发都是从一段几十行的脚本开始的拼一个 prompt调一次大模型接口拿到返回文本解析一下结束。这个模式在问答场景下确实够用但只要任务稍微复杂一点——比如需要查资料、算数、改文件、多轮确认——它立刻就会露馅。我最早做的一个自动化小工具就是这种结构用户输入一句话我把这句话塞进 prompt模型返回一段 JSON我按 JSON 里的字段去执行动作。刚开始跑得挺顺直到遇到一个需要先查天气、再根据天气决定穿什么、最后生成一段出行建议的任务。模型第一次返回的 JSON 里让我去查天气我查完把结果塞回去它又返回了一个新的 JSON 让我去查另一个城市——因为它在第一轮里把城市名理解错了。整个流程没有状态记录没有循环控制没有失败重试全靠我在外面写 if-else 硬撑。撑到第三轮我就放弃了因为代码已经变成了一团乱麻。这个经历让我意识到一个核心问题LLM 本身只是一个无状态的文本变换函数。你给它一段文本它给你一段文本。它不会记住上一轮发生了什么不会主动决定我该调用哪个工具也不会在工具报错时自己换一条路走。这些能力全部需要外部的运行时来提供。所谓 Harness Runtime就是包裹在 LLM 外面的那一层骨架。它负责维护对话历史、管理工具注册与调用、控制循环的终止条件、处理异常与重试、记录中间状态。没有这层骨架LLM 就只是一个会说话的文本生成器有了这层骨架它才变成一个能做事的 Agent。这篇文章想聊的就是这层骨架到底该怎么搭。从最简单的单次调用出发一步步推到完整的 Harness Runtime中间会经过哪些架构选择、每个选择背后的取舍是什么、实际落地时哪些地方最容易翻车。适合已经写过一两次 LLM 调用、想往 Agent 方向深入的人看也适合正在做技术选型、需要判断我到底该用多重的框架的人参考。2. 四种主流 Agent 架构的横向拆解在动手写代码之前先把架构选项理清楚。市面上关于 Agent 架构的说法很多但落到工程实现上真正有区分度的其实就四种。我按控制权在谁手里这个维度来分因为这是决定后续所有设计的关键。2.1 单次调用加后处理最轻但最容易误用这是最原始的形态。一次 API 调用模型返回文本外部代码做解析和后处理。它的本质是模型只负责生成不负责决策。适用场景非常明确分类、抽取、改写、翻译、单轮问答。这些任务的共同点是——不需要模型根据中间结果调整下一步。你给它输入它给你输出中间没有分支。我见过不少人把这种结构硬套到多步任务上做法是在 prompt 里写请你一步步思考然后输出最终答案。这其实就是早期 Chain-of-Thought 的思路。它在某些推理任务上确实有效但有个致命缺陷模型在思考过程中如果走错了路外部代码没有任何机会干预。它只能一路错到底最后给你一个看起来很有道理但实际错误的答案。提示如果你的任务需要调用外部工具查数据库、发请求、读文件单次调用结构基本可以直接排除。因为工具调用的结果必须回传给模型而单次调用没有回传这个环节。2.2 ReAct 循环让模型自己决定下一步ReAct 是 Reasoning Acting 的缩写核心思路是让模型在每一轮输出中同时给出思考和行动。行动可以是调用某个工具也可以是给出最终答案。外部运行时执行工具把结果塞回对话历史再让模型进行下一轮。这个结构的控制流是这样的把用户输入和工具描述一起发给模型模型返回一段包含 Thought / Action / Action Input 的文本运行时解析出 Action执行对应工具把工具结果作为 Observation 追加到对话历史回到第 1 步直到模型输出 Final Answer 或达到最大轮数它的优势在于灵活性模型可以根据中间结果动态决定下一步不需要你预先写死流程。缺点是不可控模型可能陷入循环、可能选错工具、可能在某个步骤上反复试错。我实测下来ReAct 在工具数量少于 10 个、任务步骤少于 8 步的场景下表现比较稳。一旦超过这个规模模型选错工具的概率会明显上升而且循环终止的判断也变得不可靠。2.3 Plan-and-Execute先规划再执行这个结构把规划和执行拆成两个阶段。第一阶段让模型生成一个完整的步骤列表第二阶段逐步执行每一步。执行过程中如果某一步失败可以选择重新规划。它解决的是 ReAct 的一个核心痛点模型在每一步都要重新理解全局目标容易迷失。先规划相当于给模型一个路线图执行时只需要关注当前这一步。但代价是灵活性下降。如果执行到第三步发现第一步的前提假设错了整个计划可能都要推翻。所以这种结构更适合步骤之间依赖关系明确、执行环境相对稳定的任务比如从指定网页抓取数据、清洗、存入数据库这种流程化作业。2.4 多 Agent 协作分工但增加通信成本当任务复杂到单个 Agent 的工具集和上下文都装不下时就会考虑拆成多个 Agent。每个 Agent 有自己的角色、工具集和上下文通过某种消息机制协作。常见的有主管-工人模式一个协调者 Agent 分配任务给多个执行者和流水线模式每个 Agent 处理一个阶段输出传给下一个。这种结构的通信开销和调试难度都是指数级上升的。我个人的经验是除非单 Agent 的上下文确实装不下否则不要轻易上多 Agent。很多看起来需要多 Agent 的场景其实用单 Agent 加更好的工具设计就能解决。架构类型控制权适用任务规模主要风险单次调用外部代码单步无法处理多步ReAct 循环模型3-8 步循环、选错工具Plan-and-Execute模型规划外部执行5-15 步计划僵化多 Agent分散15 步以上通信成本高3. Harness Runtime 的六个核心模块确定了架构方向之后接下来要解决的是运行时到底要包含哪些东西。我把一个完整的 Harness Runtime 拆成六个模块每个模块解决一个具体问题。你可以根据实际需求裁剪但理解每个模块的职责有助于你在出问题时快速定位。3.1 对话状态管理不只是存消息列表最直觉的做法是维护一个 messages 数组每轮把新消息 append 进去。但实际跑起来会发现几个问题。第一是上下文长度。对话轮数一多messages 数组会迅速膨胀很快就会超出模型的上下文窗口。这时候需要做截断或摘要。截断的策略有几种保留最近 N 轮、保留系统提示加最近 N 轮、或者对早期对话做摘要压缩。我一般用系统提示 最近 6 轮完整对话 更早内容的摘要这个组合实测在大多数任务上够用。第二是消息角色的一致性。不同模型对 role 的定义有细微差别有的要求 tool 结果必须紧跟 tool_call 消息有的允许穿插。如果状态管理没做好很容易出现消息顺序不合法的报错。第三是中间状态的持久化。如果 Agent 跑到一半进程挂了重启后能不能从断点继续这需要把状态存到外部存储文件、数据库、Redis 都行而不是只放在内存里。# 一个简化的状态管理结构 class AgentState: def __init__(self, system_prompt, max_turns10): self.messages [{role: system, content: system_prompt}] self.max_turns max_turns self.turn_count 0 self.tool_results [] # 单独记录工具调用历史便于调试 def add_user(self, content): self.messages.append({role: user, content: content}) def add_assistant(self, content, tool_callsNone): msg {role: assistant, content: content} if tool_calls: msg[tool_calls] tool_calls self.messages.append(msg) def add_tool_result(self, tool_call_id, result): self.messages.append({ role: tool, tool_call_id: tool_call_id, content: str(result) }) self.tool_results.append(result) def should_stop(self): return self.turn_count self.max_turns3.2 工具注册与调度描述比实现更重要工具系统的核心不是怎么执行工具而是怎么让模型知道有哪些工具、每个工具干什么、参数怎么填。这部分做不好模型就会频繁选错工具或填错参数。我踩过的一个坑是工具描述写得太简略。比如一个查询订单的工具描述只写了查询订单信息模型根本不知道参数该传订单号还是用户 ID。后来我把描述改成根据订单号查询订单的详细状态参数 order_id 为字符串格式的订单编号例如 ORD-20240101-001选对率立刻上去了。工具描述要包含四个要素功能说明、参数含义、参数格式、一个具体示例。这四样缺一不可。调度层面需要处理的是模型返回的 tool_call 可能有多个并行调用也可能参数格式不对比如该传数字传了字符串。运行时要做参数校验和类型转换校验失败时把错误信息回传给模型让它重试而不是直接抛异常终止。3.3 循环控制与终止条件防止无限打转ReAct 循环最怕的就是模型陷入死循环。我遇到过模型连续五轮都在调用同一个工具、传同样的参数因为它没意识到上一次的结果已经回答了问题。终止条件至少要设三层最大轮数硬性上限比如 15 轮到了就强制停止并返回当前状态重复检测如果连续两轮的工具调用和参数完全相同判定为循环中断显式终止模型输出 Final Answer 或特定终止标记时正常结束注意最大轮数不要设得太小。我一开始设了 5 轮结果很多正常任务跑到第 6 步才完成全被截断了。后来改成 15 轮配合重复检测效果比较平衡。3.4 错误处理与重试区分可恢复和不可恢复工具执行失败是常态。网络超时、参数错误、权限不足、返回格式异常各种情况都会遇到。关键是要区分哪些错误值得重试哪些应该直接放弃。网络超时、临时性服务不可用这类可以重试一般重试 2-3 次每次间隔递增。参数错误、权限不足这类重试也没用应该把错误信息回传给模型让它换个方式或换个工具。还有一种情况是模型返回的 tool_call 格式本身就不合法比如 JSON 解析失败。这时候不要直接崩溃而是把解析错误作为一条系统消息塞回去让模型重新生成。3.5 可观测性日志、追踪与回放Agent 跑起来之后最痛苦的事情是它为什么做了这个决定。没有日志你只能看到最终输出中间发生了什么完全是个黑盒。我的做法是记录三类信息每一轮的完整输入输出包括发给模型的 messages 和模型返回的原始内容、每次工具调用的参数和结果、每一轮的耗时和 token 消耗。这些信息存成结构化的 JSON 日志出问题时可以完整回放整个决策链路。更进一步可以做 trace给每个任务分配一个 trace_id所有相关日志都带上这个 ID方便在日志系统里串联。3.6 安全边界工具权限与输出过滤Agent 能调用工具就意味着它能对真实世界产生影响。发邮件、改文件、调支付接口这些操作一旦被模型误触发后果可能很严重。基本的安全措施包括工具分级只读工具和写入工具分开写入工具需要额外确认、参数白名单比如文件路径限制在特定目录下、输出过滤模型生成的最终输出在返回给用户前做敏感信息检查。我一般会把工具分成三档绿色只读直接执行、黄色有副作用但可撤销执行前记录、红色不可逆操作必须人工确认。这个分级在配置里写死不依赖模型的判断。4. 从零搭一个最小可用的 Runtime理论讲完了接下来动手。我以一个能查资料、能算数、能读写本地文件的 Agent 为例把 Runtime 的骨架搭出来。代码用 Python 写模型接口用通用的 chat completions 格式你可以替换成任何兼容的模型服务。4.1 工具层的定义与注册先定义工具。每个工具是一个函数加一份描述。描述用 JSON Schema 格式因为大多数模型接口都认这个格式。import json import math # 工具注册表 TOOL_REGISTRY {} def register_tool(name, description, parameters): def decorator(func): TOOL_REGISTRY[name] { function: func, schema: { type: function, function: { name: name, description: description, parameters: parameters } } } return func return decorator register_tool( namecalculator, description执行数学计算支持加减乘除和幂运算。参数 expression 为数学表达式字符串例如 2 3 * 4, parameters{ type: object, properties: { expression: { type: string, description: 要计算的数学表达式例如 2 3 * 4 } }, required: [expression] } ) def calculator(expression): # 只允许安全字符防止注入 allowed set(0123456789-*/.() ) if not set(expression).issubset(allowed): return {error: 表达式包含不允许的字符} try: result eval(expression, {__builtins__: {}}, {}) return {result: result} except Exception as e: return {error: f计算失败: {str(e)}}这里有个细节值得说eval本身是不安全的所以我做了字符白名单校验并且把__builtins__置空。生产环境更稳妥的做法是用ast.literal_eval或者专门的表达式解析库但作为演示白名单校验已经能挡住大部分注入尝试。4.2 主循环的实现骨架主循环负责串联调用模型 → 解析工具调用 → 执行工具 → 回传结果这个链条。import time class AgentRuntime: def __init__(self, llm_client, model, max_turns15): self.llm llm_client self.model model self.max_turns max_turns self.state None self.trace [] def run(self, user_input, system_prompt): self.state AgentState(system_prompt, self.max_turns) self.state.add_user(user_input) for turn in range(self.max_turns): self.state.turn_count turn 1 start time.time() # 1. 调用模型 response self.llm.chat( modelself.model, messagesself.state.messages, tools[t[schema] for t in TOOL_REGISTRY.values()] ) elapsed time.time() - start # 2. 记录 trace self.trace.append({ turn: turn 1, elapsed: round(elapsed, 2), response: response }) # 3. 判断是否有工具调用 tool_calls response.get(tool_calls, []) if not tool_calls: # 没有工具调用视为最终答案 return response[content] # 4. 执行工具 self.state.add_assistant(response.get(content, ), tool_calls) for call in tool_calls: result self._execute_tool(call) self.state.add_tool_result(call[id], result) # 5. 重复检测 if self._is_repeating(): return 检测到重复调用已中断 return 达到最大轮数任务未完成 def _execute_tool(self, call): name call[function][name] try: args json.loads(call[function][arguments]) except json.JSONDecodeError: return {error: 参数解析失败请检查 JSON 格式} if name not in TOOL_REGISTRY: return {error: f未知工具: {name}} try: return TOOL_REGISTRY[name][function](**args) except TypeError as e: return {error: f参数不匹配: {str(e)}} except Exception as e: return {error: f执行异常: {str(e)}} def _is_repeating(self): if len(self.trace) 2: return False last self.trace[-1][response].get(tool_calls, []) prev self.trace[-2][response].get(tool_calls, []) if not last or not prev: return False return json.dumps(last, sort_keysTrue) json.dumps(prev, sort_keysTrue)这段代码里_execute_tool的错误处理是重点。参数解析失败、工具不存在、参数不匹配、执行异常四种情况分别返回不同的错误信息这些信息会作为 tool 结果回传给模型模型看到之后有机会自我修正。4.3 把工具结果喂回模型的格式细节不同模型接口对 tool 结果的格式要求不完全一样。有的要求 tool 消息必须带tool_call_id有的要求name字段有的两者都要。我建议在状态管理里统一处理对外暴露一个简单的add_tool_result方法内部根据配置的模型类型做格式转换。还有一个容易忽略的点工具返回的内容长度。如果工具返回了一大段文本比如读了一个大文件直接塞回 messages 会迅速撑爆上下文。我的做法是在工具层做截断超过一定长度比如 2000 字符就截断并加提示内容过长已截断如需完整内容请指定范围。4.4 一次完整任务的执行追踪假设用户输入是帮我算一下 123 乘以 456然后把这个结果加上 789。第一轮模型返回 tool_callcalculator({expression: 123 * 456})。运行时执行得到 56088回传。第二轮模型看到 56088返回 tool_callcalculator({expression: 56088 789})。执行得到 56877回传。第三轮模型返回最终答案123 乘以 456 等于 56088加上 789 等于 56877。整个 trace 里记录了三次模型调用、两次工具执行每步的耗时和 token 消耗都清清楚楚。如果哪一步出了问题回放 trace 就能定位。5. 实测中暴露的五个典型问题与修复上面那套骨架跑通之后我在实际使用中陆续遇到了一些问题。这些问题在文档里很少提到但几乎每个做 Agent 的人都会碰到。5.1 模型假装调用了工具这是最隐蔽的一个坑。模型在返回的 content 里写了一段看起来像工具调用的文本比如我将调用 calculator 计算 22但实际上并没有生成真正的 tool_calls 字段。运行时看到没有 tool_calls就把它当成最终答案返回了用户看到的就是一句我将调用...然后没有下文。修复方法是在系统提示里明确要求必须通过 tool_calls 字段调用工具不要在 content 里描述调用意图。另外可以在运行时加一层检测如果 content 里出现了工具名加括号的模式但 tool_calls 为空就判定为异常重新提示模型。5.2 参数类型不匹配导致的静默失败模型有时候会把数字参数传成字符串比如{expression: 123}而不是{expression: 123}。如果工具函数没有做类型检查可能会静默失败或者产生意外结果。我的做法是在工具注册时声明参数类型运行时做一次类型转换尝试。转换失败就返回错误信息让模型重试。这个转换逻辑要小心不能把abc强行转成数字只能做安全的转换。5.3 上下文膨胀导致的性能下降对话轮数一多每轮发给模型的 messages 越来越长响应时间明显变慢token 消耗也上去了。我实测过一个任务跑到第 10 轮时单次调用的输入 token 是第 1 轮的 8 倍。解决办法是滑动窗口加摘要。保留系统提示和最近 5 轮完整对话更早的对话用一段摘要代替。摘要可以在轮数达到阈值时触发一次额外的模型调用生成虽然多花一次调用但后续每轮省下的 token 远超过这个成本。5.4 工具执行超时拖垮整个循环有个工具是调用外部接口查数据正常情况下 1 秒内返回但偶尔会遇到接口卡住一卡就是几十秒。整个 Agent 循环就停在那里等用户体验极差。修复是给每个工具执行加超时控制。Python 里可以用concurrent.futures的ThreadPoolExecutor配合future.result(timeoutN)实现。超时后返回一个执行超时的错误让模型决定是重试还是换工具。from concurrent.futures import ThreadPoolExecutor, TimeoutError _executor ThreadPoolExecutor(max_workers4) def execute_with_timeout(func, args, timeout10): future _executor.submit(func, **args) try: return future.result(timeouttimeout) except TimeoutError: return {error: f工具执行超时{timeout}秒}5.5 多轮之后模型忘记了原始目标这是 ReAct 结构的固有问题。跑到第 7、8 轮时模型容易被中间的工具结果带偏忘记了用户最初要的是什么。缓解方法有两个一是在每轮调用时把原始用户输入重新强调一遍放在 messages 靠前的位置或者作为系统提示的一部分二是在系统提示里明确要求每完成一步后回顾原始目标确认当前步骤是否在正确的路径上。我试过在每轮的消息末尾加一句提醒用户的原始需求是 XXX效果比较明显但会增加 token 消耗。折中方案是每 3 轮加一次提醒。6. 架构选型的判断清单与演进路径聊了这么多实现细节最后回到选型问题。很多人纠结的是我到底该用多重的框架。我的建议是不要一上来就追求完整而是根据任务复杂度逐步演进。6.1 什么情况下不需要 Harness如果任务满足以下所有条件单次调用就够了不需要任何 Harness输入输出都是纯文本不需要调用外部工具不需要多轮交互一次问答就能解决不需要根据中间结果调整策略分类、抽取、翻译、改写、单轮问答这些都属于这一类。硬上 Harness 只会增加复杂度和故障点。6.2 什么时候该上 ReAct当任务需要调用外部工具且调用哪个工具、传什么参数需要根据中间结果决定时ReAct 是最自然的选择。典型场景查资料加总结、多步计算、需要读写文件的任务。判断标准很简单如果你在写代码时发现需要根据上一步的结果来决定下一步做什么那就该上 ReAct。6.3 什么时候需要 Plan-and-Execute当任务步骤超过 8 步或者步骤之间的依赖关系比较复杂时ReAct 容易迷失。这时候先规划再执行会更稳。典型场景数据流水线、多阶段报告生成、需要按固定流程走但每步参数动态确定的任务。6.4 多 Agent 的适用边界只有当单个 Agent 的工具集超过 15 个、或者上下文确实装不下所有信息时才考虑拆多 Agent。拆的时候要明确每个 Agent 的职责边界和通信协议否则调试成本会高到让你怀疑人生。任务特征推荐架构理由单步文本处理单次调用无需状态和工具3-8 步工具调用ReAct灵活且实现简单8-15 步流程化Plan-and-Execute减少迷失工具超 15 个多 Agent上下文和工具集拆分6.5 演进路径从脚本到 Runtime 的四个阶段我自己的演进路径是这样的供参考第一阶段写一个单文件脚本硬编码工具调用逻辑。这个阶段的目标是验证模型能不能完成这个任务。第二阶段把工具抽出来做成注册表主循环用 ReAct 结构。这个阶段的目标是让新增工具变得容易。第三阶段加上状态持久化、错误重试、日志追踪。这个阶段的目标是让 Agent 能在生产环境稳定跑。第四阶段根据任务特点选择 Plan-and-Execute 或多 Agent并做性能优化上下文压缩、并行工具调用。这个阶段的目标是处理更复杂的任务。每个阶段之间不需要推倒重来都是在前一阶段基础上叠加。关键是不要跳过验证阶段直接上重型架构那样出了问题你根本不知道是哪一层的问题。我在实际项目里最大的体会是Agent 的难点从来不在模型本身而在模型外面那层运行时。模型的能力是给定的但运行时怎么设计、怎么处理异常、怎么控制循环、怎么记录状态这些才是决定一个 Agent 能不能真正干活的关键。把 Runtime 的每个模块职责理清楚遇到问题时能快速定位到是哪一层出了状况比盲目换模型或换框架有用得多。
阅读完成 · 觉得有帮助?
咨询建站