Agent Harness是什么？用公司类比拆解12个核心组件

Agent Harness 这个词，我在好几个技术群里都被问起过。很多人一看到 Agent 就以为 Agent Harness 是什么高深莫测的智能体框架，接着就开始讨论 Agent Harness 和 Agent 到底谁取代谁。说实话，我第一次接触这个词的时候也绕了很久的弯路，后来我用一个非常生活化的类比把它彻底想通了：Agent 是员工，Agent Harness 是让员工能稳定产出价值的公司中后台系统。这篇文章不讲花哨概念，只做一件事：从这套"公司系统"的类比出发，一步步推导出 Agent Harness 里最常见的 12 个组件，每个组件我都会说清它是干什么的、怎么落技术实现、有哪些常规文档里不会写的坑。适合正在做 Agent 应用开发、正在从 demo 走向线上生产、或者想在技术方案评审时把架构讲清楚的朋友。

1. 内容整体设计与思路拆解

1.1 先搞清楚：Harness 到底是什么，和 Agent 有什么本质区别

先解决最大的困惑。搜索"harness 到底是啥"的人很多，因为它本身是英语里的"马具、挽具、安全带"，后来被软件工程借用来表示"测试夹具"或"执行框架"。放到智能体语境里，Agent 指的是那个能感知、决策、行动的智能主体，它通常由大模型驱动，配上工具和记忆；而 Harness 指的是把 Agent 装进去、管起来、让它持续可靠运行的那一层外壳。

我用公司类比解释：Agent 是一个刚入职的员工，能干、有潜力，但裸奔着进来是不行的。他需要工位、电脑、门禁权限、项目资料、汇报机制、考核制度、应急预案。把他一个人丢在仓库里，他再聪明也发挥不出来。Agent Harness 就是这一整套"公司基础设施"，它不负责某个任务具体怎么思考，它负责让 Agent 的思考有输入、有输出、有约束、有记录、有兜底。

两者的关系不是替代，而是分层。你写一个循环，调大模型、调工具，那是 Agent 逻辑；你把这个逻辑放进一个可重入的会话容器，加上日志追踪、权限控制、配置管理、错误重试，这就是 Harness。工程上真正难、也真正体现平台价值的，往往是后者。

1.2 用"新员工入职公司"这个类比，直接推演出整套架构

我习惯在设计复杂系统前先找一个生活化类比做骨架。如果你把一个基于大模型的 Agent 应用看作一家公司，那么 Agent 是员工，Agent Harness 就是公司为支撑员工工作而建立的整套机制。顺着"员工从入职到干活到复盘"这条线走，组件会自然浮现：

这 12 个组件不是一个完美框架必须全上，但它们合起来确实覆盖了 Agent 应用从开发到上线的完整生命周期。接下来逐个拆。

1.3 为什么要组件化，而不是在一个大函数里写逻辑

很多人会问：我就写一个脚本，循环调几次大模型，加个 tools 参数，不也能跑吗？何必分成 12 个组件？能跑和能交付是两回事。单体脚本适用于个人实验，一旦要上线，问题立刻暴露：换模型要改代码、prompt 变了没法回溯、线上出问题不知道卡在哪一步、工具权限没人管、并发一高 session 串上下文、模型抽风一次整个任务直接失败。

组件化的本质是关注点分离 + 可替换性。模型网关把供应商隔离掉，你从 GPT 换到 Claude，只改配置不改业务代码；编排引擎把流程控制权收走，你可以从单循环平滑升级到工作流；观测组件让你不用打断点就能复盘任何一次线上决策。我做的每个 Agent 项目，哪怕再小，至少会先切出模型网关、工具注册、追踪日志这三层，成本很低，收益立竿见影。

2. 12 个组件逐个拆解：从比喻到落地

2.1 第一组：执行底座类组件，解决"Agent 怎么跑起来"

组件 1：运行时容器（Runtime Container）

类比是工位和电脑，公司不可能让员工拿着自己的电脑在公司核心系统上乱跑。技术上一个 Agent 应用也必须跑在受控环境里：加载模型配置、初始化工具链、创建独立会话上下文、分配并发资源、处理超时取消。我做过一个最典型的教训：早期把 Agent 跑在无状态的 Web Worker 里，每个请求进来都临时 new 一个实例，结果模型调用的连接池频繁重建，响应时间翻倍。后来把运行时容器做成进程级单例，内部按 session_id 隔离状态，连接复用，性能立刻正常。运行时容器还需要处理执行生命周期：start、step、pause、resume、stop。这意味着你的 Agent 循环不能是裸 for，至少要能被外部信号打断、被调用方查询进度。

组件 2：模型网关（Model Gateway）

员工不可能直接认识全公司所有部门的人，他需要通过行政通道去沟通。对大模型来说，网关就是这个通道。它统一封装 OpenAI、Claude、Gemini、本地部署模型，对外暴露一个相近的 chat 接口；在网关内部做密钥管理、模型路由、超时重试、流量限制、token 统计。没有网关的最大问题是散：项目里到处是openai.ChatCompletion.create，换一次供应商要搜全仓替换。网关层我会建议至少做三件事：第一，统一返回结构，把消息内容、工具调用、token 用量、原始元数据全包进去；第二，在此层做成本累加，按会话维度记账；第三，对模型返回做一次基础解析校验，格式不对早失败，别让脏数据流入后续流程。

组件 3：提示与上下文装配器（Prompt & Context Builder）

员工能不能干好活，往往取决于他拿到的资料和指令是否清晰。对应到技术上，就是把 system prompt、用户原始输入、检索到的知识片段、历史会话摘要、工具函数定义，拼装成符合模型接口要求的消息列表。这个组件听上去简单，实际很容易写烂。我见过太多项目把 prompt 拼接散落在业务代码里，一会儿 f-string，一会儿 append，最终同一个 prompt 在不同调用点生成的消息结构都不一样。规范做法是：所有模板集中声明，变量显式注入，输出前统一 dump 一次做检查。我调试 prompt 最常用的就是打印法：把最终组装出来的 messages 完整打印出来，人工看一遍到底模型收到了什么。很多"模型不听话"的问题，根源其实是装配器拼错了结构。

组件 4：工具注册表与执行器（Tool Registry & Executor）

公司里员工要调用财务系统、申请资源、查询数据库，不会自己乱写接口。技术上也一样，Agent 的所有外部能力必须通过工具注册表暴露。工具注册表维护一份结构化清单：工具名、描述、参数 JSON Schema、执行函数、权限等级。模型通过 function calling 选择工具后，执行器负责解析参数、做类型校验、执行函数、把结果包装成消息回填给模型。这里有个关键点：工具描述就是给模型看的提示词，描述写得暧昧，模型就会选错。我踩过的一个坑是给搜索工具写"search anything you need"，结果模型什么都调它，把大量简单问答也丢进搜索再等结果，又慢又贵。改成具体描述后，比如"仅在上游订单未找到时调用，用于查询订单状态"，准确率立刻提升。另外，工具执行结果必须做截断和摘要，否则一个超长 JSON 回灌上下文，一次就把窗口撑爆。

2.2 第二组：流程控制类组件，解决"Agent 怎么决策和记忆"

组件 5：编排引擎（Orchestrator / Planner）

员工需要一个主管来安排任务顺序和目标拆解，Agent 也需要编排引擎来维护"下一步做什么"。编排引擎是 Agent Harness 里最核心也最容易过度设计的部分。常见形态有三种：ReAct 单循环（模型每轮决定是调工具还是给最终回复）、Plan-and-Execute（先产出计划再逐步执行）、Graph/状态机（把流程显式建模为节点和边）。我的建议非常务实：前期永远从单循环开始，把决策权交给模型，你只需要维护 messages 列表；跑通后再根据稳定性需求，把固定流程抽取成 Graph 节点。这里我要多说一句：现在很多人一上来就搭复杂图编排，一个客服机器人画几十个节点，最后维护成本远超收益。真实业务里 90% 的流程是线性的，加上两三个条件分支就足够，复杂编排只留给强流程控制的场景。

组件 6：记忆与状态管理（Memory & State Manager）

员工不可能每次开会都让客户重新介绍一遍背景，公司有客户档案，员工有工作笔记。对应到 Agent Harness，记忆分三层。第一层是会话内短期记忆，也就是消息队列里已经存在的上下文；第二层是会话摘要，当上下文接近模型窗口上限时，把早期对话压缩成摘要；第三层是长期记忆，跨会话存储用户偏好、历史订单、知识条目，通常用向量数据库或普通 KV 存储。长期记忆不要做成只读，一定要提供"写入"机制，而且写入动作应该显式化。我在项目里会把save_memory、retrieve_memory也注册成工具，让 Agent 在需要沉淀信息时主动调用，比硬编码在每次对话后自动存效果更可控。还要注意记忆内容的隐私边界，敏感信息尽量存引用 ID，展示和检索时再做解析不到处落明文。

组件 7：事件总线与通信（Event Bus / Messaging）

公司内部沟通靠 IM 和邮件，系统与系统之间靠接口和消息队列。Agent Harness 里的事件总线有两个职责：一是内部解耦，让模型网关、日志、限流、回调都通过事件钩子扩展，而不是硬编码在主循环里；二是支撑多 Agent 协作和外部系统集成。比如你有一个客服 Agent 和一个质检 Agent，质检 Agent 就需要订阅客服 Agent 发出的"工单已生成"事件。实现时我建议在最底层定义一组标准事件：agent_start、tool_call、tool_result、model_response、agent_end，每个事件带 session_id、时间戳、载荷。基于这组事件，你后续接日志、接监控、接审计都非常顺手。这里有个易忽略的问题：事件消费者要尽量幂等，因为同一事件可能因为重试被投递多次。

2.3 第三组：治理保障类组件，解决"Agent 能不能放心上路"

组件 8：可观测性与追踪（Observability / Tracing）

员工干了什么要有日报周报，系统跑了什么要有日志链路。Agent 应用的可观测性比传统后端更重要，因为它的每一步都是不确定性决策。线上用户问了一句"我的订单呢"，模型可能先查了用户表，又查了订单表，中间还调错一次工具，最后给了个错误答复。没有追踪你根本无从还原。我在生产项目里强烈建议把每次模型调用、每个工具调用的入参出参、token 数、耗时全部记录到 trace，并支持按 session_id 回放。这不是性能开销的问题，是能不能调试的问题。很多团队问为什么 Agent 线上效果不稳定，先别急着换模型，把 trace 打开看看是不是工具结果格式变了、prompt 版本被改了、上下文被截断。

组件 9：安全与权限护栏（Security & Guardrails）

公司不会把财务审批权限放给实习生，Agent 也不能拥有无限权限。Guardrails 要做三层：输入侧防注入，输出侧做内容合规过滤，动作侧做权限与审批控制。动作侧是最容易被忽略的，模型自己调用删除接口和建议人工确认，这两个性质完全不同。我给 Tool Executor 增加了一个"敏感操作二次确认"机制：当工具标记为敏感时，执行器不直接调用，而是先返回"需要 manager 审批"的占位消息，然后通过事件总线通知人工系统，等审批通过后继续。这个机制可以抽象成一个叫request_approval的工具，模型在流程里显式调用它，比外围拦截更自然。还要提醒一点：guardrail 的判断逻辑本身不要只靠大模型，关键动作的资格校验必须用确定性代码检查。

组件 10：配置与特征开关（Config & Feature Flags）

公司制度会定期更新，员工手册会出新版，Agent 的配置也一样。配置组件管理的包括：使用哪个模型、temperature 参数、system prompt 版本、工具集合开关、数据库连接、生产环境地址。我见过太多团队直接在代码里改 prompt 再发布，结果线上问题来回横跳。规范做法是配置中心化，YAML 或 JSON 定义 schema，启动时用 pydantic 校验；prompt 模板和模型参数纳入 git 管理，通过配置系统的版本号发布。我还会给每个配置项加一个"作用域"字段，比如某个新工具只在内部测试流量中启用，这就是最简单的灰度发布。切记不要在业务代码里写一堆 if env == "production" 的分支，配置应该走统一加载，而不是散落环境判断。

组件 11：评估与回放（Evaluation / Replay）

员工干得好不好要看 KPI，Agent 改得好不好要靠评测集。Agent 的评测和传统模型的离线评测不同，它不能只看最终答案，还要看中间决策过程：有没有在不需要工具时乱调工具、有没有生成危险内容、有没有及时放弃错误分支。我的做法是先录 50 条真实线上会话作为回归集，每次改动 prompt 或组件后批量回放，对比"最终回答质量"和"关键中间动作符合率"。初期不需要搞多复杂的 LLM-as-judge 评分体系，先把"同样的输入，改动前后的输出和步骤差异"人工看一遍，就能拦住大部分回归问题。等积累到一定量级，再引入自动化打分器。

组件 12：容错与自愈（Resilience / Self-healing）

员工遇到突发状况要有应急预案，Agent 也一样。LLM 调用会超时、工具会报错、模型偶尔会输出非法 JSON，Harness 必须做四件事：超时控制、重试退避、熔断降级、检查点恢复。经验之谈：LLM 网关返回 429、500、网络错误时要做指数退避重试，但返回 400 这类参数错误时千万不要重试，重试只会浪费钱。工具执行错误要区分可重试和不可重试，比如查询类工具可以重试，但"已发送邮件"这类副作用操作不能盲目重试，需要补偿机制。最后一定加一个全局兜底：Agent 连续决策失败 N 次或者达到最大步数时，把会话转交人工处理，这比在系统里继续硬撑优雅得多。

3. 实操过程与核心环节实现：从零搭一个最小可用的 Harness

讲完组件，我们落地一个最小骨架。我以一个"售后工单回复助手"为场景：用户提问，Agent 决定是否查询订单状态，最后给出回复。这个 demo 不会依赖具体大模型 SDK，把核心流程写清楚，你换成任何模型网关都能跑。

3.1 先设计组件交互流程

场景链路是：用户输入 -> 上下装装配 -> 模型决策 -> 如果调工具就执行 -> 结果回填 -> 再次模型决策 -> 输出最终答案。按照这个链路，我会拆出四个核心类：ModelGateway、ToolRegistry、SessionMemory、Runner，外加一个Config和一个日志钩子。

from dataclasses import dataclass, field from typing import Any, Optional import json, time, uuid, logging log = logging.getLogger("harness") class Config: def __init__(self, config_dict: dict): # 所有配置集中在 dict 中，用 pydantic 校验更佳 self.model = config_dict["model"] self.temperature = config_dict.get("temperature", 0.2) self.max_steps = config_dict.get("max_steps", 8) self.system_prompt = config_dict["system_prompt"] self.model_provider = config_dict.get("model_provider", "openai")

Config 就这样，实际项目里建议用 YAML 文件 + pydantic 校验。关键是模型名、系统提示词、最大步数全部走配置，不改代码。

class ModelGateway: """模型网关：统一模型调用入口，记录 token 与耗时。""" def __init__(self, config: Config): self.config = config def chat(self, messages: list, tools: Optional[list] = None) -> dict: # 真实项目中这里按 model_provider 路由到具体 SDK # 我这里用一个假的响应模拟模型返回，方便你跑通流程 time.sleep(0.1) # 模拟模型决定调用查询工具的响应 if tools and "查询订单状态" in messages[-1]["content"]: return { "message": { "role": "assistant", "content": None, "tool_calls": [ {"id": "call_1", "function": {"name": "get_order_status", "arguments": json.dumps({"order_id": "A1024"})}} ] }, "token_usage": {"prompt": 120, "completion": 30} } return { "message": { "role": "assistant", "content": "您的订单 A1024 已发货，预计明天到达。", "tool_calls": None }, "token_usage": {"prompt": 100, "completion": 25} }

这个 ModelGateway 是演示用的 mock，真实场景里替换成调用 OpenAI SDK 的代码即可。注意到它内部统一返回了message和token_usage两个字段，业务代码不需要关心模型厂商差异。

class ToolRegistry: """工具注册表：注册工具、暴露 schema、执行工具。""" def __init__(self): self._tools = {} def register(self, name: str, description: str, schema: dict, func: callable): self._tools[name] = {"description": description, "schema": schema, "func": func} def schemas(self) -> list: out = [] for name, info in self._tools.items(): out.append({"type": "function", "function": { "name": name, "description": info["description"], "parameters": info["schema"] }}) return out def execute(self, name: str, arguments: dict) -> dict: if name not in self._tools: return {"error": f"unknown tool {name}"} try: result = self._tools[name]["func"](**arguments) return {"result": result} except Exception as e: return {"error": str(e)}

工具注册表是 Harness 里最好用也最容易扩展的部分。我习惯把权限检查放在 execute 内部最开始，先校验调用方 session 是否有权执行该工具，再进行调用。

def get_order_status(order_id: str) -> str: # 真实项目里这里查订单服务 return f"订单 {order_id} 状态：已发货，物流单号 SF1234567890" tool_registry = ToolRegistry() tool_registry.register( name="get_order_status", description="查询订单状态。仅当用户询问订单物流、订单状态时使用。", schema={"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]}, func=get_order_status )

注意工具描述："仅当用户询问订单物流、订单状态时使用"。我前面强调过，描述要具体，要能帮助模型做选择。你写"查询订单"可能也行，但加上使用场景限制，误调率会低很多。

class SessionMemory: """会话记忆：按 session_id 保存消息列表，并做长度截断。""" def __init__(self, max_messages: int = 20): self.sessions = {} self.max_messages = max_messages def get(self, session_id: str) -> list: return self.sessions.setdefault(session_id, []) def append(self, session_id: str, message: dict): self.sessions.setdefault(session_id, []).append(message) # 超出长度直接丢最早的消息，生产环境建议替换成摘要压缩 if len(self.sessions[session_id]) > self.max_messages: self.sessions[session_id] = self.sessions[session_id][-self.max_messages:]

SessionMemory 承担了记忆组件的最小职责。这个实现的截断策略非常粗暴：超长直接丢弃。真实产品里这里应该是"滑动窗口 + 摘要"的组合策略，早期先把能存能丢做对。注意 session 用字典存在进程内，多实例部署必须换成 Redis 这类共享存储。

class Runner: """编排引擎：主循环 + 事件钩子 + 兜底容错。""" def __init__(self, gateway: ModelGateway, tools: ToolRegistry, memory: SessionMemory, config: Config): self.gateway = gateway self.tools = tools self.memory = memory self.config = config def _build_messages(self, session_id: str, user_input: str) -> list: system = {"role": "system", "content": self.config.system_prompt} history = self.memory.get(session_id) if not history or history[0].get("role") == "system": history = [system] + history return history + [{"role": "user", "content": user_input}] def run(self, user_input: str, session_id: Optional[str] = None) -> str: session_id = session_id or uuid.uuid4().hex messages = self._build_messages(session_id, user_input) log.info("[agent_start] session=%s input=%s", session_id, user_input) for step in range(self.config.max_steps): response = self.gateway.chat(messages, tools=self.tools.schemas()) message = response["message"] messages.append(message) self.memory.append(session_id, message) # 收集事件：这里本来会发给事件总线，先落到日志 log.info("[model_response] step=%d tokens=%s", step, response["token_usage"]) tool_calls = message.get("tool_calls") if not tool_calls: return message["content"] for tc in tool_calls: fn_name = tc["function"]["name"] args = json.loads(tc["function"]["arguments"] or "{}") log.info("[tool_call] name=%s args=%s", fn_name, args) tool_result = self.tools.execute(fn_name, args) log.info("[tool_result] name=%s result=%s", fn_name, tool_result) messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": json.dumps(tool_result, ensure_ascii=False) }) # 达到最大步数，转人工兜底 log.warning("[agent_max_steps] session=%s steps=%d", session_id, self.config.max_steps) return "REACHED_MAX_STEPS_NEED_HUMAN_HANDOVER"

Runner 是整套 Harness 的核心编排。它做的事情很简单：循环、调模型、看是否有工具调用、执行工具、回填结果。关键点在于每个环节都打了日志，这相当于内置了一个最小可观测层。如果你想扩展事件总线，把日志调用替换成event_bus.publish(...)即可。

config = Config({ "model": "gpt-4o-mini", "system_prompt": "你是售后客服助理。当用户询问订单状态时，你必须调用 get_order_status 工具查询后再回复。不要编造订单信息。", "max_steps": 8 }) gateway = ModelGateway(config) memory = SessionMemory() runner = Runner(gateway, tool_registry, memory, config) reply = runner.run("我的订单 A1024 发货了吗？没有物流信息。", session_id="test_1") print(reply)

这就是一个能跑通的最小 Harness。你跑一遍会看到日志完整记录了：任务开始、模型决策、工具调用、工具结果、最终回复。这个骨架只有 150 行左右，但它已经包含 12 个组件中的模型网关、工具注册表、上下文装配、记忆、编排、日志、兜底。其他组件（配置、权限、追踪、评测等）都是在这个骨架上加挂的中间件，不是另起炉灶。

实际项目里我会把这个骨架放进一个harness/包，每新增一个组件就往 Runner 的循环里插入一个钩子。比如加入 Guardrails，就在tool_result回到 messages 之前加一个校验函数；加入 Trace，就在log.info位置接入 OTel Span。结构清爽，扩展不伤筋动骨。

4. 常见问题与排查技巧实录

4.1 模型调用了工具，但工具参数永远是错的

这是最气人的问题之一。现象是 Agent 明明知道该用get_order_status，却总是把order_id传成None或者传一个找不到的字段。排查顺序：第一步看工具 schema 是不是写标准了，required有没有漏；第二步看工具描述里有没有给出参数示例，建议在描述里直接写"order_id 形如 A1024，不要传完整句子"；第三步检查工具执行器是不是把模型的 arguments 字符串正确解析了，有些框架里参数是 JSON 字符串嵌套字符串，解析一次不够。我见过特别多"参数错误"其实是解析层的问题，不是模型的问题。

4.2 对话轮次一多，模型开始答非所问

这是上下文管理的问题。messages 列表无限增长，旧消息把关键信息挤出了窗口，或者早期系统提示词被大量工具输出稀释了。我的排查手法是先用 trace 看线上那几轮失败会话的最终 messages 结构，检查窗口被什么占用了。大概率是工具返回了大段 JSON。解决方式二级叠加：一是工具结果统一做截断和摘要，超过 500 字符要压缩；二是在 SessionMemory 里做摘要，比如每 6 轮把早期对话压缩成一段 summary。很多团队把记忆组件做成向量数据库就觉得万事大吉，实际上向量数据库管的是长期记忆，短期窗口管理才是高频痛点。

4.3 如何判断一次改动是变好还是变坏

Agent 应用是非确定性的，同样的输入可能输出不同结果，导致回归测试很难做。我推荐先搭"回放对比"工具：把线上录制的 N 条请求回放到新旧两个配置下，生成前后两次 trace，然后在关键节点上做字段比对：工具是否调用、工具参数是否合理、最终回复是否满足约束、有没有触发 guardrail。你可以先用人工快速过 20 条，再逐步把比对的规则自动化。当初期没有评测集时，这条路径是最不费力的质量保障手段，比你写一百条 prompt 试炼集有用得多。

4.4 组件越加越多，系统慢得没法用

Harness 组件化不等于要一次性全上。很多人照着架构图把所有组件都实现一遍，结果每次请求要过配置检查、权限校验、日志采集、指标上报、评测抽样、模型网关、工具解析，光中间件耗时就要几百毫秒。我的经验是组件分层：核心路径组件（模型网关、上下文装配、工具注册、编排、基础日志）必须有，进阶治理组件（评测、事件总线、自动化护栏）在第二阶段上，而且尽量用异步或旁路方式，不要阻塞主链路。做技术方案不是比谁组件全，是比谁能在合适规模下保持迭代速度。

4.5 常见问题速查表

写到最后的一点经验

回头看这篇文章，用一个"员工和公司"的类比把 Agent Harness 的整套架构拆开，确实比直接念概念好记忆得多。我个人在实际设计中还有个习惯：先把要做的 Agent 应用画成"公司组织架构图"，员工是哪个大模型，行政部门对应哪些组件、风控对应哪些组件，画完再映射成技术模块。这样既能查漏补缺，也能在评审会上让非技术同事听懂。最后送你一句踩坑换来的忠告：不要在第一天就追求 12 个组件全部到位，先把模型网关、工具注册、日志追踪这三块焊死，剩下的按业务复杂度慢慢长出来。组件化是为了让你在复杂面前不失控，不是为了让你在简单面前显得专业。