文章总结： 本文介绍不依赖框架、仅用OpenAISDK加函数注册表与while循环手写约150行AgentHarness的实战方法。核心是四层架构：循环控制、上下文管理、工具注册与LLM推理，强调可控性与可观测性。提供完整可运行代码示例，并给出从极简到生产的升级路径，建议按需叠加持久化、权限控制等能力。 综合评分： 88 文章分类： AI安全,安全开发,安全建设
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] + "\n...[结果已截断]"
六、完整实战：一个可跑的 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"\n=== 最终回答 ===\n{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 行为的完全可见和完全可控。
极简不是目的，可控才是目的。框架可以帮你写更快的第一版，但只有你自己写过一遍循环，你才知道框架到底替你藏了什么——而那些藏起来的东西，往往就是线上事故的来源。
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：威胁情报Z分析 Z Z《Agent Harness 实战：用极简工具构建可控的 AI Agent》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论