Harness & Hermes：从模型能力到可交付智能体的工程闭环

原创

用户12339161

发布于 2026-10-12 17:49:41

1450

Hermes 是大脑，Harness 是躯体。没有 Harness 的 Hermes 只是一个会说话的模型，没有 Hermes 的 Harness 只是一个空转的调度器。两者结合，才能把概率性智能装进确定性流程，交付真正的 AI 数字员工。

一、概念对齐：各司其职

Hermes 指 Nous Research 的 Hermes 系列模型，核心优势是函数调用与结构化输出。它在训练阶段就强化了 <tool_call> 协议解析，输出格式远比通用模型稳定，适合作为智能体的推理核心。

Harness 指智能体运行时（Agent Harness），负责模型之外的一切：消息历史管理、工具注册与执行、轮次控制、超时重试、权限校验、审计日志、人工兜底。

用一个比喻：Hermes 是刚毕业的天才研究生，脑子聪明但没工牌、没电脑、不知道内部流程；Harness 是给他配齐工牌、电脑、内部 Wiki 和审批权限的入职系统。Model + Harness = Agent，缺一不可。

二、Harness 的四项核心职责

一个生产级 Harness 必须解决四个问题：

消息循环：管理 Human → AI → Tool → AI 的接力式消息链。
工具执行：解析 Hermes 输出的 <tool_call>，执行并回填 <tool_response>。
权限边界：白名单控制可调用工具，高风险操作强制审批。
可观测性：记录每步工具调用、耗时、token 用量，支持追溯。

三、代码实现：最小可用的 Harness

以下代码展示 Harness 如何驱动 Hermes 完成一次工具调用闭环：

import json, re, time, uuid
import requests

class HermesHarness:
    def __init__(self, api_url: str, model: str, tools: dict,
                 max_steps: int = 6, timeout: int = 30):
        self.api_url = api_url
        self.model = model
        self.tools = tools          # {name: callable}
        self.max_steps = max_steps
        self.timeout = timeout
        self.audit = []

    def _build_system(self) -> str:
        specs = [{"name": n, "description": f.__doc__ or ""}
                 for n, f in self.tools.items()]
        return (
            "你在 <tools> 中定义工具。\n"
            f"<tools>{json.dumps(specs, ensure_ascii=False)}</tools>\n"
            "需要时输出 <tool_call>{\"name\":...,\"arguments\":{...}}</tool_call>"
        )

    def _call_model(self, messages: list) -> str:
        resp = requests.post(self.api_url, json={
            "model": self.model,
            "messages": messages,
            "temperature": 0,
            "timeout": self.timeout,
        }, timeout=self.timeout)
        return resp.json()["choices"][0]["message"]["content"]

    def _parse_tool_call(self, text: str):
        m = re.search(r"<tool_call>(.*?)</tool_call>", text, re.S)
        return json.loads(m.group(1)) if m else None

    def run(self, user_input: str) -> dict:
        messages = [
            {"role": "system", "content": self._build_system()},
            {"role": "user", "content": user_input},
        ]
        for step in range(self.max_steps):
            output = self._call_model(messages)
            call = self._parse_tool_call(output)

            if not call:
                return {"answer": output, "audit": self.audit}

            name = call["name"]
            args = call.get("arguments", {})

            # 权限校验：白名单外的工具直接拒绝
            if name not in self.tools:
                result = {"error": f"工具 {name} 未授权"}
            else:
                try:
                    result = self.tools[name](**args)
                except Exception as e:
                    result = {"error": str(e)}

            self.audit.append({
                "step": step, "tool": name, "args": args,
                "result": result, "ts": time.time()
            })

            messages.append({"role": "assistant", "content": output})
            messages.append({"role": "user", "content":
                f"<tool_response>{json.dumps(result, ensure_ascii=False)}</tool_response>"})

        return {"answer": "达到最大步数，转人工", "audit": self.audit}

使用示例：

def get_weather(city: str) -> dict:
    """查询城市天气"""
    return {"city": city, "temp": 22, "condition": "sunny"}

def create_ticket(order_id: str, reason: str) -> dict:
    """创建工单"""
    return {"ticket_id": f"T-{order_id}", "status": "created",
            "idempotency_key": str(uuid.uuid4())}

harness = HermesHarness(
    api_url="http://localhost:8000/v1/chat/completions",
    model="hermes-3-llama-3.1-70b",
    tools={"get_weather": get_weather, "create_ticket": create_ticket},
)
result = harness.run("北京天气怎么样？如果下雨帮我建个工单")
print(json.dumps(result, ensure_ascii=False, indent=2))

四、生产化增强：从能跑到好用

上面的骨架解决了"能用"，生产环境还需四个增强：

第一，幂等键。 所有写操作必须携带幂等键，超时重试不产生重复副作用。可在 Harness 层自动为高风险工具注入：

HIGH_RISK = {"create_ticket", "send_email", "update_record"}

def _ensure_idempotency(self, name: str, args: dict) -> dict:
    if name in HIGH_RISK and "idempotency_key" not in args:
        args["idempotency_key"] = str(uuid.uuid4())
    return args

第二，人工审批。 高风险操作不直接执行，而是挂起等待确认：

def _requires_approval(self, name: str) -> bool:
    return name in {"update_record", "send_email"}

# 在 run() 中：
if self._requires_approval(name):
    return {"status": "pending_approval", "tool": name,
            "args": args, "audit": self.audit}

第三，预算熔断。 按 token 用量和步数双重限制：

def _check_budget(self, used_tokens: int, step: int):
    if used_tokens > 100_000 or step >= self.max_steps:
        raise RuntimeError("预算超限，熔断转人工")

第四，并行工具执行。 当 Hermes 一次返回多个互不依赖的 <tool_call> 时，用 asyncio.gather 并发执行，把总耗时从 t1+t2+t3 压到 max(t1,t2,t3)。

五、多智能体扩展

当单 Agent 不够用时，Harness 可升级为编排器：Planner 拆解任务，多个 Worker 并行执行，Reviewer 校验结果。每个子 Agent 拥有独立的 Harness 实例与工具白名单，只有最终摘要返回父级。

class MultiAgentHarness:
    def __init__(self, planner: HermesHarness,
                 workers: dict[str, HermesHarness]):
        self.planner = planner
        self.workers = workers

    def run(self, task: str):
        plan = self.planner.run(f"拆解任务为子任务JSON：{task}")
        results = {}
        for subtask in json.loads(plan["answer"]):
            worker = self.workers[subtask["agent"]]
            results[subtask["goal"]] = worker.run(subtask["goal"])
        return results

六、工程要点

协议稳定：Hermes 的 <tool_call> 格式比自由文本可靠，但仍需正则容错。
权限前置：白名单在 Harness 层执行，不依赖模型自觉。
审计完整：每步记录工具名、参数、结果、耗时，支持回放。
人工兜底：低置信度、轮次超限、高风险操作全部转人工。
成本可控：缓存相同问题，限制 max_tokens，监控 token 用量。

总结：Harness 管流程，Hermes 管决策，工具管执行。三者职责清晰、边界明确，才能把概率性智能装进确定性系统。代码不到百行，却包含了消息循环、工具执行、权限校验、审计日志的完整闭环。把 Harness 和 Hermes 串起来，AI 才真正从"对话助手"进化为"数字员工"。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

模型测试