智能体开发实战02｜Harness工程入门

上篇我们搭了第一个 Agent，60 行跑通了「思考→调用工具→生成回答」的循环。

但说实话，如果你真把它拿到实际场景里去跑几天，很快就会发现一堆让人头疼的问题：

• 搜索做到一半，API 超时了，整个程序直接崩掉。
• 模型像着了魔一样连续调用工具 10 次就是不回答，眼瞅着 Token 哗哗地烧光。
• 搜索工具返回了一堆乱码，Agent 愣了几秒，然后开始一本正经地胡说八道。
• 跑了一个小时，你完全不知道中间发生了什么。

这些问题的根子在哪？说白了，你的 Agent 目前只有一套“应用程序”级别的代码——调用模型、返回结果，仅此而已。它完全不具备“操作系统”级别的能力，来管理资源、处理异常、约束边界。

而这个“操作系统”，就是今天要聊的 Harness。

这篇是系列的第二步，也是第二阶段的开门之作。我们会从零开始，给上篇那个有点“裸奔”的 Agent 装上 Harness，让它真正能扛住真实场景的考验。

一、Harness 到底管什么

Harness 不是什么玄乎的复杂框架，它本质上是一整套让 Agent 能稳定运行的工程机制。我们可以用一个类比来理解：

没有 Harness 的 Agent，就像一台没有操作系统的裸机——电是通上了，但随时可能蓝屏给你看。

二、从第一行代码看问题

不妨回过头来看看，上篇 Agent 的核心逻辑长什么样：

# ======================================================================# [问题] 上篇代码没有任何保护措施# ======================================================================response = client.chat.completions.create(...) # 网络超时？直接崩msg = response.choices[0].message # response 为 None？崩溃if msg.tool_calls:result = web_search(args["query"])# 搜索抛异常？直接崩# 没有最大循环限制 -> 模型可以无限调工具，Token 烧光# 没有日志 -> 跑完不知道它做了什么# 没有降级 -> 工具挂了，整个 Agent 也挂了

在实际生产环境中，上面这每一行代码，都可能是程序的命门。

三、Harness 核心组件实现

我们先从最常用的 5 个组件入手，一个一个把它们加到 Agent 身上。

1. 重试机制（Retry）

网络抖动、API 限流，这些事儿在现实世界就跟吃饭喝水一样平常。重试不能傻乎乎地连打，必须带上退避策略，不然容易把服务器给打炸了。

import timeimport random# ================================================================# 组件 1: 重试机制（Retry）-- 网络抖动？给我再试一次# ================================================================def retry_on_fail(func, max_retries=3, base_delay=1):"""带指数退避（Exponential Backoff）的重试包装器参数:func:要执行的函数max_retries: 最大重试次数（默认 3 次）base_delay:初始等待秒数（每次翻倍）效果:失败 1 次等 1s，失败 2 次等 2s，失败 3 次等 4s ..."""for attempt in range(max_retries):try:return func()except Exception as e:if attempt == max_retries - 1:raisedelay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)print(f"[attempt {attempt + 1}/{max_retries}] "f"failed: {e}, retry in {delay:.1f}s...")time.sleep(delay)

2. 超时控制（Timeout）

模型或者工具可能突然“失联”，我们不能一直干等着。比较常用的做法是借助 signal 或 Thread 来实现超时中断。

import signal# ================================================================# 组件 2: 超时控制（Timeout）-- 不能无限等下去# ================================================================class TimeoutError(Exception):"""自定义超时异常"""passdef _timeout_handler(signum, frame):"""SIGALRM 信号处理函数：直接抛异常"""raise TimeoutError("操作超时")def with_timeout(func, timeout_sec=30):"""给任意函数加超时保护（基于 SIGALRM）参数:func:要执行的函数timeout_sec: 超时秒数（默认 30s）返回:函数执行结果，超时则返回 None"""signal.signal(signal.SIGALRM, _timeout_handler)signal.alarm(timeout_sec)try:return func()except TimeoutError:print(f"[timeout] operation exceeded {timeout_sec}s, terminated")return Nonefinally:signal.alarm(0)

3. 最大步数限制（Max Steps）

这是最容易被忽略的一个陷阱。模型有时候会陷入“思考→调工具→再思考→再调工具”的循环里出不来，必须给它设一道硬上限。

# ================================================================# 组件 3: 最大步数限制（Max Steps）-- 防死循环的保险丝# ================================================================def run_with_step_limit(agent_func, max_steps=5):"""限制 Agent 最大执行步数，防止陷入死循环参数:agent_func: Agent 单步执行函数，返回 {"done": bool, ...}max_steps:最大允许步数（默认 5 步）返回:正常完成 -> Agent 结果超出限制 -> 错误字典 + 部分结果"""last_result = Nonefor step in range(max_steps):last_result = agent_func()if last_result and last_result.get("done"):return last_resultprint(f"[max_steps] reached limit ({max_steps}), force terminating")return {"error": "max_steps_exceeded","partial": last_result,}

4. 日志追踪（Logging）

如果连 Agent 刚才干了什么都不知道，那基本就等于没有调试能力。每一步操作都应该留下结构化的记录。

import jsonfrom datetime import datetime# ================================================================# 组件 4: 日志追踪（Logging）-- 每一步都看得见# ================================================================class AgentLogger:"""结构化 Agent 日志器，记录每一步操作"""def __init__(self):self.logs = []def log(self, event, detail):"""记录一条日志。event:事件类型（如 TOOL_CALL / MODEL_RESPONSE / TIMEOUT）detail: 事件详情（字符串或字典）"""entry = {"timestamp": datetime.now().isoformat(),"event": event,"detail": detail,}self.logs.append(entry)if isinstance(detail, dict):detail_str = json.dumps(detail, ensure_ascii=False)else:detail_str = str(detail)print(f"[log] [{event}] {detail_str[:100]}")def dump(self):"""导出完整日志（列表形式）"""return self.logs# ===== 使用示例 =====logger = AgentLogger()logger.log("TOOL_CALL",{"tool": "web_search", "query": "AI news 2026"},)logger.log("MODEL_RESPONSE",{"tokens": 256, "finish_reason": "stop"},)

5. 降级策略（Fallback）

核心工具挂了，不等于整个 Agent 就要跟着陪葬。准备一个 B 计划至关重要：

# ================================================================# 组件 5: 降级策略（Fallback）-- 工具挂了，Agent 不能挂# ================================================================def safe_tool_call(func, fallback_result=None):"""工具调用安全包装：重试 -> 失败 -> 降级参数:func:要执行的工具函数fallback_result: 降级时的替代返回值（可选）执行链:func -> 重试（最多 3 次）-> 降级 -> 返回友好提示"""try:return retry_on_fail(func)except Exception as e:print(f"[fallback] tool call failed ({e}), using fallback")if fallback_result is not None:return fallback_resultreturn "（工具暂不可用，请稍后再试）"

四、组装：带 Harness 的 Agent

现在，我们把上面这 5 个组件全部装进上篇的 Agent 里，升级成一个真正有“操作系统”底气的版本：

# ======================================================================#Harness Agent -- 完整可运行版本（带操作系统）# ======================================================================#比上篇多了：重试、超时、步数限制、日志追踪、降级策略#复制后替换 API Key 即可运行# ======================================================================import jsonimport randomimport requestsimport signalimport timefrom datetime import datetimefrom openai import OpenAI# ======================================================================#第一步：配置# ======================================================================client = OpenAI(api_key="your-api-key-here",base_url="https://api.deepseek.com/v1",)MODEL = "deepseek-chat"# ======================================================================#第二步：装配 Harness 组件# ======================================================================# ---------- 组件 1: 日志追踪 ----------class AgentLogger:"""结构化日志器，记录 Agent 每一步操作"""def __init__(self):self.logs = []def log(self, event, detail):entry = {"timestamp": datetime.now().isoformat(),"event": event,"detail": detail,}self.logs.append(entry)print(f"[log] [{event}] {str(detail)[:80]}")def dump(self):"""导出完整日志（列表形式）"""return self.logsharness_log = AgentLogger()# ---------- 组件 2: 重试机制 ----------def with_retry(func, retries=3, delay=1):"""指数退避重试：失败后等待时间呈指数增长"""for i in range(retries):try:return func()except Exception as e:if i == retries - 1:raisewait = delay * (2 ** i) + random.uniform(0, 0.5)harness_log.log("RETRY",f"attempt {i + 1}: {e}, retry in {wait:.1f}s",)time.sleep(wait)# ---------- 组件 3: 超时控制 ----------class TimeoutError(Exception):"""自定义超时异常"""passdef _signal_handler(signum, frame):"""SIGALRM 信号处理函数：直接抛异常"""raise TimeoutError("操作超时")def with_timeout(func, sec=30):"""SIGALRM 超时保护"""signal.signal(signal.SIGALRM, _signal_handler)signal.alarm(sec)try:return func()except TimeoutError:harness_log.log("TIMEOUT", f"exceeded {sec}s")return Nonefinally:signal.alarm(0)# ---------- 组件 4: 安全调用（重试 + 超时 + 降级） ----------def safe_call(func, fallback="（工具暂不可用）"):"""安全调用链：超时 -> 重试 -> 降级"""try:return with_timeout(lambda: with_retry(func))except Exception as e:harness_log.log("FALLBACK", str(e))return fallback# ======================================================================#第三步：定义 Agent 可用工具# ======================================================================TOOLS = [{"type": "function","function": {"name": "web_search","description": "搜索互联网获取最新信息","parameters": {"type": "object","properties": {"query": {"type": "string","description": "搜索关键词",},},"required": ["query"],},},},]def web_search(query):"""执行互联网搜索，返回前 3 条结果摘要"""url = "https://api.bocha.cn/v1/web-search?query=" + queryresp = requests.get(url, timeout=15)results = resp.json().get("results", [])lines = [f"{r['title']}: {r['snippet']}"for r in results[:3]]return "n".join(lines)# ======================================================================#第四步：组装 Harness Agent# ======================================================================def run_agent(user_input, max_steps=5):"""带 Harness 的 Agent 主循环。工作流程:用户输入-> 模型思考-> 是否调工具？-> 是 -> 安全调用 -> 再思考-> 否 -> 返回最终答案"""messages = [{"role": "system","content": "你是一个资讯助手，使用 web_search 获取最新信息。",},{"role": "user","content": user_input,},]for step in range(max_steps):harness_log.log("STEP", f"step {step + 1}/{max_steps}")# --- 安全调用模型（自动重试 + 超时） ---response = safe_call(lambda: client.chat.completions.create(model=MODEL,messages=messages,tools=TOOLS,tool_choice="auto",))if response is None:return json.dumps({"error": "模型调用失败"},ensure_ascii=False,)msg = response.choices[0].messagemessages.append(msg)harness_log.log("FINISH_REASON",msg.finish_reason or "none",)if not msg.tool_calls:return msg.contentfor tc in msg.tool_calls:harness_log.log("TOOL_CALL",f"{tc.function.name}({tc.function.arguments})",)if tc.function.name == "web_search":args = json.loads(tc.function.arguments)result = safe_call(lambda: web_search(args["query"]),fallback="搜索暂时不可用，请稍后再试",)messages.append({"role": "tool","tool_call_id": tc.id,"content": result,})harness_log.log("MAX_STEPS", f"reached {max_steps} steps limit")return "已达到最大执行步数，请简化问题后重试。"# ======================================================================#第五步：运行！# ======================================================================if __name__ == "__main__":answer = run_agent("今天AI圈有什么大新闻？")print()print("=" * 50)print("Agent 回答:")print(answer)print()print("=" * 50)print("Harness 运行日志:")print(json.dumps(harness_log.dump(), ensure_ascii=False, indent=2))

运行效果

正常运行时，输出一切如常。但一旦遇上 API 超时、工具报错、模型死循环这些突发状况，Harness 会立刻介入：

• 自动重试（你会在日志里看到 RETRY）
• 超时了直接中断（TIMEOUT，并返回一个降级结果）
• 超过 5 步自动终止（MAX_STEPS）
• 每一步都有日志，方便事后回溯

这就是所谓的“操作系统”级别的保障——你的 Agent 再也不会因为一次小小的网络抖动，就落得个全盘崩溃的下场。

五、Harness 组件全景图

到了这一步，你的 Agent 已经成功从“裸机”升级到了“带 OS 的电脑”——不会再动不动就蓝屏了。

六、写在最后

很多团队其实都踩过这个坑：挑选模型的时候，恨不得选最强的；但写 Harness 的时候，却用的是最弱的。结果呢？最强模型跑在最差的系统上，真用起来的体验，可能还不如一个弱模型配上好 Harness。

这个道理值得记住：Harness 直接决定了 Agent 在生产环境下能活多久，而模型决定了它回答的质量上限。两者就像是硬币的两面，缺一不可。

今天，我们给 Agent 加装了重试、超时、步数限制、降级策略和日志追踪这 5 个组件。下一篇，我们会解决那个让每个开发者都肉疼的问题——Token 成本，咱们第 03 篇见。

• 第03篇: 上下文工程——吃透 Context，Token 成本降 80%
• 第04篇: 工具调用——让 Agent 真正『动手干活』
• 第05篇: Agent 记忆系统——从『转头就忘』到『过目不忘』

回看整个旅程：第一阶段我们用 60 行代码跑通了第一个 Agent，帮大家建立了直观的感受。从这篇开始，我们正式进入了真正的工程世界——每个组件都是生产级别的，每一行代码都能直接拿来用。

今天这篇文章的代码总共不到 120 行，却给 Agent 装上了一个真正的“操作系统”。这些组件不是那种用一次就扔的东西——你会反复用到它们，直到它们变成你的肌肉记忆。

当然，如果你想在生产环境中偷懒，直接用 LangGraph 或 CrewAI 这些现成框架也行。但理解了底层原理，出了问题你才知道该去哪里修，这比什么都重要。

下一篇：上下文工程，带你吃透 Context，把 Token 成本打下来！