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

上篇文章中，我们用60行代码搭建了第一个Agent，顺利跑通了「思考→调用工具→生成回答」的完整循环。

然而，将其部署到实际环境中运行几天后，一系列棘手问题便会接踵而至：

• 搜索执行到一半，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需要随之崩溃。准备一套备用方案至关重要：

# ================================================================# 组件 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-searchquery=" + 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成本降下来！