Agent Harness 实战：用极简工具构建可控的 AI Agent
原创
Z
Z
威胁情报Z分析
2026年10月2日 08:57
广东
在小说阅读器读本章
去阅读
在公众号小说中沉浸阅读
大模型本身只是一个接收文本、吐出文本的推理引擎。真正让它变成\"能干活的 Agent\"的，是外面那层薄薄的运行时外壳——Agent Harness：负责把用户请求组装成上下文、调用模型、解析工具调用、执行工具、把结果塞回去、再问一次。
很多人一上来就上 LangChain、LlamaIndex，结果被抽象层淹没，出了问题不知道哪一层在做什么。本文走相反的路：不依赖任何 Agent 框架，只用 OpenAI SDK + 一个函数注册表 + 一个 while 循环，手写一个约 150 行的可用 Agent，看清每一个字节在流动。
一、为什么要自己写 Harness
框架替你做了三件事：循环、工具抽象、上下文管理。但这三件事各自都不复杂，封装之后反而带来三个代价：
- 不可观测：框架内部偷偷截断了历史、改了消息格式，你拿到的 trace 和模型实际看到的不一致。
- 不可控：想改一步重试策略、想加一个断点调试，要穿透几层继承关系。
- 不可靠：版本升级偷偷改默认行为，线上 Agent 行为漂移。
极简 Harness 的原则是：每一行胶水代码都你自己写，每一条消息都你自己拼，出了问题你知道在哪打 print。框架等你的 Agent 真的需要并发、持久化、多人协作时再引入不迟。
二、四层架构：每层只做一件事
整个 Harness 可以拆成四层，层与层之间通过普通的 Python 数据结构（dict 形式的 message 列表）通信，没有任何基类、没有任何装饰器。
Agent Harness 四层架构：每层只做一件事
不引入框架，用约 200 行胶水代码把 LLM、工具、上下文和循环粘合成一个可控 Agent
这四层的职责边界非常清晰：Loop Controller 只控制\"什么时候停\"，Context Manager 只控制\"送什么进去\"，Tool Registry 只控制\"函数怎么被调起来\"，LLM Core 只负责推理。任何一层想跨层干活，都是复杂度的开始。
三、Agent Loop：整个系统的心脏
Agent 的运行本质上是一个 while 循环。它的逻辑简单到可以画成一张图：
关键在那个判断节点：LLM 返回的 finish\_reason 是 tool_calls 还是 stop。前者意味着模型想让你干活，后者意味着它认为任务完成了。绝大多数\"Agent 不干活\"或\"Agent 停不下来\"的问题，都出在这两个分支的处理上。
四、极简工具：函数即工具
不需要 BaseTool 类、不需要 @tool 装饰器。一个工具就是一个普通 Python 函数，加上一段 JSON Schema 描述它的参数。注册就是把它们放进一个 dict：
import jsonfrom typing importCallable, Dict# 工具函数本身就是普通函数defcalculator(expression: str) -> str: \"\"\"安全地计算一个数学表达式\"\"\" allowed = set(\"0123456789+-*/(). \") ifnotset(expression) <= allowed: return\"错误：表达式包含非法字符\" try: returnstr(eval(expression, {\"__builtins__\": {}}, {})) except Exception as e: returnf\"计算失败：{e}\"defget_time() -> str: \"\"\"返回当前日期时间\"\"\" from datetime import datetime return datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\")# Schema 直接写给 OpenAI function calling 格式TOOLS = [ { \"type\": \"function\", \"function\": { \"name\": \"calculator\", \"description\": \"计算数学表达式，例如 1+2*3\", \"parameters\": { \"type\": \"object\", \"properties\": { \"expression\": {\"type\": \"string\", \"description\": \"数学表达式\"} }, \"required\": [\"expression\"] } } }, { \"type\": \"function\", \"function\": { \"name\": \"get_time\", \"description\": \"获取当前日期时间\", \"parameters\": {\"type\": \"object\", \"properties\": {}} } }]# 名字到函数的映射TOOL_MAP: Dict[str, Callable] = { \"calculator\": calculator, \"get_time\": get_time,}
这里没有任何魔法。Schema 是 OpenAI API 原生要求的格式，函数就是普通函数，TOOL_MAP 负责把模型说的名字映射回真实函数。想加新工具？写一个函数、加一条 Schema、往 dict 里塞一项，三步。
五、上下文管理：别让窗口炸掉
Agent 跑久了最大的敌人不是模型笨，是上下文窗口爆炸——工具返回的日志太长、历史轮次太多，把系统提示和用户指令挤出了窗口。极简策略是按固定比例预留：
对应的代码非常短：每轮把消息列表送给模型之前，先检查总 token 数，超过阈值就从最旧的一条非 system 消息开始删。工具结果单独截断到 2K tokens，避免一个 cat 大文件 吃掉整个窗口。
MAX_TOOL_RESULT_TOKENS = 2000RESERVE_OUTPUT_TOKENS = 2000deftrim_messages(messages: list, model_context_limit: int = 120000) -> list: \"\"\"从前往后裁剪非 system 消息，直到总 token 数在预算内\"\"\" budget = model_context_limit - RESERVE_OUTPUT_TOKENS # 简单按字符数粗估 token（中文约 1.5 字/token，英文约 4 字符/token） defest_tokens(msgs): returnsum(len(m.get(\"content\", \"\") or\"\") // 3for m in msgs) # system 消息永远保留 system_msgs = [m for m in messages if m[\"role\"] == \"system\"] other_msgs = [m for m in messages if m[\"role\"] != \"system\"] # 从最旧的开始删 while other_msgs and est_tokens(system_msgs + other_msgs) > budget: other_msgs.pop(0) return system_msgs + other_msgsdeftrim_tool_result(result: str) -> str: iflen(result) < MAX_TOOL_RESULT_TOKENS * 3: return result return result[: MAX_TOOL_RESULT_TOKENS * 3] + \"\
...[结果已截断]\"
六、完整实战：一个可跑的 Agent
把上面三块拼起来，就是整个 Harness。下面这段代码可以直接跑（需要 openai>=1.0 和一个 API key）
from openai import OpenAIimport jsonclient = OpenAI() # 读环境变量 OPENAI_API_KEYMODEL = \"gpt-4o-mini\"MAX_STEPS = 10# ---- 把第四节的 TOOLS 和 TOOL_MAP 放进来 ----# from tool_registry import TOOLS, TOOL_MAPdefrun_agent(user_query: str): messages = [ {\"role\": \"system\", \"content\": \"你是一个可以调用工具的助手。需要计算或查时间时调用工具。\"}, {\"role\": \"user\", \"content\": user_query}, ] for step inrange(MAX_STEPS): # 1. 裁剪上下文 messages = trim_messages(messages) # 2. 调模型 resp = client.chat.completions.create( model=MODEL, messages=messages, tools=TOOLS, ) msg = resp.choices[0].message messages.append(msg.model_dump(exclude_none=True)) # 3. 判断是否要调工具 ifnot msg.tool_calls: print(f\"\
=== 最终回答 ===\
{msg.content}\") return msg.content # 4. 执行每个工具调用 for tc in msg.tool_calls: name = tc.function.name args = json.loads(tc.function.arguments or\"{}\") print(f\"[step {step}] 调用 {name}({args})\") result = TOOL_MAP[name](**args) result = trim_tool_result(str(result)) messages.append({ \"role\": \"tool\", \"tool_call_id\": tc.id, \"content\": result, }) print(\"达到最大步数，强制停止\")if __name__ == \"__main__\": run_agent(\"现在几点？帮我算一下 (128+57)*3 等于多少\")
跑起来你会看到类似这样的输出
[step 0] 调用 get_time({})[step 1] 调用 calculator({'expression': '(128+57)*3'})=== 最终回答 ===现在是 2026-09-3021:30:15。(128+57)*3 = 555。
整个 Agent 没有任何继承、没有任何配置对象、没有任何中间件。你想加日志？在第 4 步 print。想加重试？包一层 try。想换模型？改一个常量。
七、从极简到生产：什么时候该加东西
上面这个版本适合脚本、个人项目和调试。真要上生产，下面这些是按需叠加的，而不是一上来就全要
| 能力 | 极简版怎么做 | 什么时候需要升级 |
|---|---|---|
| 持久化 | messages 列表在内存里，进程退出就没了 | 需要跨会话记忆 / 多轮对话中断恢复时，落 SQLite 或 Redis |
| 工具权限 | 所有函数都能被模型调用，无审批 | 工具涉及写文件、发邮件、调钱时，加 dry-run 或人工确认 |
| 错误恢复 | 工具抛异常直接崩 | 把异常字符串作为 tool result 回灌，让模型自己重试或换路径 |
| 并行工具调用 | 串行执行 tool\_calls 列表 | 模型一次返回多个无依赖调用时用 asyncio.gather 并行 |
| 观察性 | print 到 stdout | 接入 LangSmith / OpenTelemetry，记录每轮 token 数和延迟 |
| 结构化输出 | 模型自由文本回答 | 下游要消费结果时，用 response\_format 强制 JSON Schema |
八、小结
Agent Harness 的本质不是什么复杂系统，就是一个 while 循环加三个字典：消息列表、工具映射、Schema 数组。把这三个东西捏在手里，你就拥有了对 Agent 行为的完全可见和完全可控。
极简不是目的，可控才是目的。框架可以帮你写更快的第一版，但只有你自己写过一遍循环，你才知道框架到底替你藏了什么——而那些藏起来的东西，往往就是线上事故的来源。
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论