别再迷信 Workflow 了真正的 Agent 该有自己的操作系统一、开篇为什么我们需要 Harness Engineering你有没有这种感觉——用 Coze / Dify 搭的 Workflow一开始跑得挺爽一旦需求稍微飘一点就开始各种兜不住用户问了个没预设到的问题节点直接断路想动态切数据源不好意思流程写死了想加个如果失败就换个策略的逻辑画布上开始缠绕成毛线球说白了Workflow 是给确定性问题准备的餐刀而现实世界里全是不确定性的硬骨头。于是就有了 Harness Engineering 这个概念——用工程化手段让 AI Agent 可靠、持续地完成任务。它不是让你去调 Prompt而是让你去搭一套Agent 的操作系统。二、Agent Harness 的 6 大基础模块一个能上生产环境的 Harness内核至少要包含这 6 个模块模块中文职责Loop主循环调用 LLM、分发工具、判断终止。Agent 的主执行流Tool工具集LLM 能调用的所有能力Bash/Read/Write/Edit…Context上下文管理System Prompt 历史 RAG 工具结果负责压缩与窗口控制Environment沙箱环境文件系统、网络、权限、工作区隔离Memory记忆层跨轮次 / 跨会话的状态持久化短期 长期Observability可观察性日志、Trace、评估指标、回放、评测这 6 个模块是最小内核。少了任何一个Agent 要么跑不久要么跑不稳要么跑完不知道为什么。但光有内核还不够。现代 Agent 系统普遍要加一个高级编排模块Sub-Agents子智能体—— 不是底层必选但几乎是标配。三、Sub-Agent让 Agent 学会分身术3.1 一个真实痛点想象你要做一个行业技术调研报告。如果用固定 Workflow你大概会这么画问题 → 抽行业关键词 → 抽技术关键词 → 搜索 → 分析 → 生成报告问题来了反爬了怎么办中途要切数据源搜到假信息怎么办得有校验 Agent想顺便找到行业大佬联系方式这不是预设流程里的节点调研报告写一半发现方向错了Workflow好的我崩了这时候你需要的不是画布而是分工。3.2 主从 Agent 架构┌──────────────────┐ │ 主 Agent │ │ (规划 调度) │ └────────┬─────────┘ │ task(prompt) ┌──────────────┼──────────────┬─────────────┐ ▼ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ 检索Agent│ │ 摘要Agent│ │ 校验Agent│ │ 报告Agent│ │ 独立上下文│ │ 独立上下文│ │ 独立上下文│ │ 独立上下文│ └─────────┘ └─────────┘ └─────────┘ └─────────┘核心价值就两点职责清晰—— 一个 Agent 干一件专业的事上下文隔离—— 子 Agent 像子进程跑完只把结果交上来垃圾信息不会污染主 Agent 的上下文主 Agent 的上下文是战略会议室子 Agent 的上下文是作战前线。谁在前线捡到的泥土别带进会议室。3.3 为什么是 Agent不是 Workflow维度WorkflowMulti-Agent (Sub-Agents)路径预设写死自主拆解按需调用应变无法应对不确定根据中间结果动态调整分支场景简单、固定任务需求模糊、有未知问题容错弱强可换 Agent 重试一句话Workflow 是地铁线路图Agent 是出租车司机。四、手撸一个 Sub-Agent 版 Harness光说不练假把式。下面用 Python OpenAI SDK从零构建一个支持 Sub-Agent 的最小 Harness。完整代码可跑直接抄作业。4.1 环境 常量Python 中的常量约定import os import json import subprocess from pathlib import Path from openai import OpenAI from dotenv import load_dotenv load_dotenv(overrideTrue) # 工作目录Agent 被授权的沙箱根目录 WORKDIR Path.cwd() MODEL os.getenv(DEEPSEEK_MODEL) client OpenAI( base_urlos.getenv(DEEPSEEK_BASE_URL), api_keyos.getenv(DEEPSEEK_API_KEY), )Python 语法解析Python 没有真正的常量。像WORKDIR这种全大写的变量是一种约定俗成的命名规范——告诉其他开发者这是常量别改它。但语言层面并不禁止你修改这就是脚本语言和 Java 的区别。Path.cwd()返回当前工作目录的Path对象。相比老式的os.getcwd()返回字符串Path对象自带一堆链式方法比如/、.parent、.exists()这也是 Python 里面向对象化IO 的体现。load_dotenv(overrideTrue)overrideTrue意味着.env文件里的值会覆盖系统环境变量。如果不加这个参数系统已有的环境变量优先级更高——调试时踩过一次坑明明改了.env却一直读旧值就是这里。4.2 Environment 模块路径沙箱这是Environment的第一道防线任何文件操作都不能逃出工作目录。def safe_path(p: str) - Path: 返回安全路径超出工作目录直接抛异常 path (WORKDIR / p).resolve() if not path.is_relative_to(WORKDIR): raise ValueError(fPath escapes workspace: {p}) return path逐行解析path (WORKDIR / p).resolve()WORKDIR / p里的/不是除法而是pathlib.Path重载的运算符。这是 Python 少数几个运算符重载的场景之一效果等价于os.path.join(WORKDIR, p)。设计者的初衷让路径拼接像拼接 URL 一样自然。.resolve()会做两件事① 把相对路径转成绝对路径② 解析..和软链接。这一步是防逃逸的关键——如果不 resolve../../etc/passwd这种路径检测会失效。if not path.is_relative_to(WORKDIR): raise ValueError(fPath escapes workspace: {p})is_relative_to是 Python 3.9 新增的方法判断当前路径是否在目标路径之下。比字符串startswith更安全——避免/workspace和/workspace-evil这种前缀误判。raise ValueError(...)主动抛异常而不是返回None。这是 Python 的哲学错误不应该被静默吞掉。调用方必须显式处理否则整个 Agent 会挂——但挂比静默出错强。类型注解说明def safe_path(p: str) - Path:p: str表示入参是字符串- Path表示返回值是Path对象。Python 的类型注解不强制——你可以传 int 进去运行时也不报错除非逻辑本身炸了。它的作用是给 IDE、mypy 和读代码的人看的。这也是 Python 弱类型脚本语言最直白的体现。4.3 Tool 模块4 个基础工具工具 1Bashdef run_bash(command: str) - str: dangerous [rm -rf /, sudo, shutdown, reboot, /dev/] if any(d in command for d in dangerous): return Error: Dangerous command blocked try: r subprocess.run( command, shellTrue, cwdWORKDIR, capture_outputTrue, textTrue, errorsreplace, timeout120 ) out (r.stdout r.stderr).strip() return out[:50000] if out else No output except subprocess.TimeoutExpired: return Error: Timeout (120s) except (FileNotFoundError, OSError) as e: return fError: {e}语法点 1any(d in command for d in dangerous)这一行包含两个 Python 特性生成器表达式Generator Expression(d in command for d in dangerous)会惰性判断每个危险词是否在命令里。相比列表推导式[d in command for d in dangerous]它不占内存且any()一旦遇到True就短路返回不用跑完全部。any()/all()内置函数any一真即真all一假即假。这是 Python 里最常用的两个聚合判断工具。语法点 2subprocess.run参数解析参数作用坑点shellTrue通过 shell 执行命令支持管道、重定向但有注入风险cwdWORKDIR子进程的工作目录确保命令在沙箱内运行capture_outputTrue捕获 stdout 和 stderr等价于stdoutPIPE, stderrPIPEtextTrue输出以文本形式返回否则拿到的是 bytes还得手动 decodeerrorsreplace解码错误时替换而非报错防止乱码直接崩掉 Agenttimeout120超时强制 kill超时抛TimeoutExpired异常语法点 3(r.stdout r.stderr).strip()r.stdout和r.stderr都是字符串拼接不是列表别想extend。.strip()去掉首尾空白符空格、\n、\t都算。注意out[:50000] if out else No output是 Python 的三元表达式语法是值1 if 条件 else 值2——和 C 的条件 ? 值1 : 值2逻辑一致但顺序相反读起来更英语。语法点 4except (FileNotFoundError, OSError) as e用元组一次性捕获多种异常类型很常用。as e把异常对象绑定到变量efError: {e}里的{e}会自动调用str(e)。捕获顺序有讲究Python 从上到下匹配所以更具体的异常要写在前面。工具 2/3/4读、写、改def run_read(path: str, limit: int None) - str: try: lines safe_path(path).read_text(encodingutf-8).splitlines() if limit and limit len(lines): lines lines[:limit] [f(...{len(lines) - limit} more)] return \n.join(lines)[:50000] except Exception as e: return fError: {e} def run_write(path: str, content: str) - str: try: fp safe_path(path) fp.parent.mkdir(parentsTrue, exist_okTrue) fp.write_text(content, encodingutf-8) return fWrote {len(content)} bytes except Exception as e: return fError: {e} def run_edit(path: str, old_text: str, new_text: str) - str: try: fp safe_path(path) content fp.read_text(encodingutf-8) if old_text not in content: return fError: Text not found in {path} fp.write_text(content.replace(old_text, new_text, 1), encodingutf-8) return fEdited {path} except Exception as e: return fError: {e}语法点解析limit: int None—— 默认参数为None。很多人会写if limit is None判断其实if limit就够了因为0也 falsy正好符合limit0 表示不限的语义。这是个常见的 Python 惯用法。lines[:limit] [f(...{len(lines) - limit} more)]—— 列表切片 列表拼接。[:limit]是切片语法取前limit个拼接两个列表。\n.join(lines)—— 用换行符把字符串列表拼起来。这是 Python 里最高效的字符串拼接方式比for line in lines: s line快得多因为字符串在 Python 里是不可变对象每次都会创建新字符串。fp.parent.mkdir(parentsTrue, exist_okTrue)—— 两个关键参数parentsTrue自动创建所有不存在的父目录类似mkdir -p。exist_okTrue目录已存在也不报错否则会抛FileExistsError。content.replace(old_text, new_text, 1)—— 第三个参数1表示只替换第一处。改代码时非常关键否则同一段文本多处出现就全被改了。except Exception as e—— 捕获所有异常包括你可能没预料到的。这是一种Agent 友好的写法任何错误都变成字符串返回给 LLM让模型自己决定下一步而不是让整个 Agent 进程崩掉。4.4 工具注册表区分父子工具集这是关键设计——主 Agent 比子 Agent 多一个task工具。CHILD_TOOLS [ { type: function, function: { name: bash, description: Run a shell command, parameters: { type: object, properties: {command: {type: string}}, required: [command] } } }, # ...read_file / write_file / edit_file 略 ] PARENT_TOOLS CHILD_TOOLS [ { type: function, function: { name: task, description: Spawn a subagent with fresh context. It shares the filesystem but not conversation history., parameters: { type: object, properties: { prompt: { type: string, description: Short description of the task. } }, required: [prompt] } } } ] TOOL_HANDLERS { bash: lambda **kw: run_bash(kw[command]), read_file: lambda **kw: run_read(kw[path], kw.get(limit)), write_file: lambda **kw: run_write(kw[path], kw[content]), edit_file: lambda **kw: run_edit(kw[path], kw[old_text], kw[new_text]), }语法点 1PARENT_TOOLS CHILD_TOOLS [...]拼接两个列表不是相加是合并返回一个新列表CHILD_TOOLS本身不变。这正是我们要的效果两个工具集互不影响。如果你用CHILD_TOOLS.append(...)那就把子 Agent 的工具集也污染了——这是新手最常踩的坑。语法点 2lambda **kw: run_bash(kw[command])lambda是匿名函数lambda 参数: 返回值只能写一行表达式不能有语句。**kw是任意关键字参数类似 JavaScript 的 rest 参数...kw但它收集的是字典不是数组。调用时handler(**args)又把它展开回关键字参数一来一回完成了字典 → 函数参数的映射。为什么要这么绕因为 LLM 返回的工具参数是JSONPython 里就是 dict我们要用同一个字典去调用不同签名的函数。**kw就是那个万能适配器。kw.get(limit)和kw[limit]的区别get在 key 不存在时返回None[]会抛KeyError。对于可选参数用get必填参数用[]抛错反而是好事能暴露问题。语法点 3TOOL_HANDLERS字典派发用字典代替 if-else 链是 Python 里非常经典的派发模式。工具从 4 个扩到 40 个这段代码都不用改。相比if name bash: ... elif name read_file: ...字典派发更快、更易扩展、更适合动态注册。4.5 核心Sub-Agent 的执行流SYSTEM ( fYou are a coding agent at {WORKDIR}. Use task for focused exploration or a self-contained subtask. ) SUB_SYSTEM ( fYou are a coding agent at {WORKDIR}. Complete the given task, then return a concise final answer. ) def run_subagent(prompt: str) - str: # ⚠️ 关键sub_messages 是全新的从零开始 sub_messages [{role: user, content: prompt}] msg None for _ in range(30): # 独立的 Agentic Loop response client.chat.completions.create( modelMODEL, messages[{role: system, content: SUB_SYSTEM}] sub_messages, toolsCHILD_TOOLS, max_tokens8000 ) msg response.choices[0].message sub_messages.append(msg.model_dump()) if response.choices[0].finish_reason ! tool_calls: break results [] for tool_call in msg.tool_calls: func_name tool_call.function.name args json.loads(tool_call.function.arguments) handler TOOL_HANDLERS.get(func_name) output handler(**args) if handler else fUnknown tool: {func_name} results.append({ role: tool, tool_call_id: tool_call.id, content: str(output)[:50000] }) sub_messages.extend(results) # 只把最终答案交回主 Agent中间过程全部丢弃 return msg.content or (no summary)语法点 1隐式字符串拼接SYSTEM ( fYou are a coding agent at {WORKDIR}. Use task for focused exploration or a self-contained subtask. )括号里连续写多个字符串字面量Python 会自动拼成一个。这是 CPython 的词法特性不是运行时零开销。好处是可以把长 Prompt 分行写代码更好读。注意只有字面量之间才能这么拼变量不行a b是语法错误。语法点 2for _ in range(30)_是约定俗成的我不在乎这个值的变量名。这里的循环次数我们不用只是最多跑 30 次。与 JS 的for (let i 0; ...)不同Python 里如果不用i就别写i用_更清晰。语法点 3[{role: system, content: SUB_SYSTEM}] sub_messages每次调用 LLM 都临时构造一个完整 messages 列表系统提示 当前对话。为什么不直接把 system 塞进sub_messages因为 sub_messages 会被反复 append如果塞进去下一轮就会重复一次 system 消息LLM 会困惑。这是个常见的工程细节system prompt 放外面每次拼业务消息放里面持续累积。语法点 4msg.model_dump()OpenAI SDK 返回的msg是Pydantic 对象不是普通 dict。model_dump()把它序列化成 dict。为什么要转因为下一轮调用要把它塞回messages列表——而 OpenAI API 只接受纯 dict。不转的话SDK 序列化时会出问题。记住这条链LLM 返回 Pydantic 对象 →model_dump()转 dict → append 到 messages → 下次调用的输入。语法点 5if response.choices[0].finish_reason ! tool_callsfinish_reason是 OpenAI 协议的终止信号stop正常结束模型说完了tool_calls模型要求调用工具length达到 max_tokens判断逻辑只要不是tool_calls就认为这轮任务结束跳出循环。这是 Agent Loop 的刹车。语法点 6handler(**args) if handler else fUnknown tool: {func_name}三元表达式的完整写法A if 条件 else B。handler(**args)把字典 args 展开成关键字参数。如果 args 是{command: ls}等价于handler(commandls)。这是整个 Tool Calling 机制的核心桥接LLM 输出 JSON → Python 解析成 dict →**展开成函数调用。语法点 7return msg.content or (no summary)or短路如果msg.content是真值非空字符串返回它否则返回(no summary)。相当于return msg.content if msg.content else (no summary)的简写。这个惯用法在 Python 里随处可见x or default就是x 为空则用 default。4.6 主 Agent 循环调度中心def agent_loop(messages: list): while True: response client.chat.completions.create( modelMODEL, messages[{role: system, content: SYSTEM}] messages, toolsPARENT_TOOLS, max_tokens8000 ) msg response.choices[0].message messages.append(msg.model_dump()) if response.choices[0].finish_reason ! tool_calls: return results [] if msg.tool_calls: for tool_call in msg.tool_calls: func tool_call.function args json.loads(func.arguments) if func.name task: prompt args.get(prompt, ) print(f task: {prompt[:80]}) # 启动子 Agent只拿最终答案 output run_subagent(prompt) else: handler TOOL_HANDLERS.get(func.name) output handler(**args) if handler else fUnknown tool: {func.name} results.append({ role: tool, tool_call_id: tool_call.id, content: str(output) }) messages.extend(results)语法点 1def agent_loop(messages: list)参数messages是可变对象list函数内对它的修改append、extend会影响外部。这是 Python 里最著名的可变默认参数陷阱的亲戚——这里没设默认值所以安全但如果写成def agent_loop(messages[])那个空列表会跨调用共享你要的数据就会莫名其妙累积。踩过这个坑的人一辈子都记得。语法点 2messages.extend(results)vsmessages.append(results)extend把列表的每个元素逐个加入append会把整个列表当成一个元素。比如[1,2].extend([3,4])得到[1,2,3,4]而[1,2].append([3,4])得到[1,2,[3,4]]。Agent 场景必须用extend因为每条 tool 结果都是一条独立的对话消息。语法点 3args.get(prompt, )dict.get(key, default)key 不存在时返回 default而不是抛KeyError。这里防御性编程的意义LLM 有时会输出不符合 schema 的 JSON如果不设默认值Agent 会因为KeyError直接崩。给它一个至少能继续跑。语法点 4str(output)[:50000]强制把输出转成字符串万一 handler 返回 None 或对象。[:50000]截断防止工具输出把上下文窗口撑爆。工程提醒这只是粗暴的截断生产环境应该做有损压缩或摘要见下文。语法点 5tool_call_id: tool_call.id配对OpenAI 的 tool calling 协议要求assistant 消息里的每个tool_calls[i].id必须在后续role: tool消息里原样带上tool_call_id。这是给 LLM 的配对索引——告诉它这条结果对应我刚刚哪个工具请求。漏了或写错API 直接 400。这是工具调用最容易踩的坑之一。4.7 启动入口ANSI 彩色提示符if __name__ __main__: print(Subagent - fresh messages, final text returns) print(Enter a question, press Enter to send. Type q to quit.\n) history [] while True: try: query input(\033[36ms06 \033[0m) except (EOFError, KeyboardInterrupt): break if query.strip().lower() in (q, exit, ): break history.append({role: user, content: query}) agent_loop(history)语法点 1if __name__ __main__:Python 的入口惯例。当文件被直接运行时__name__是__main__当文件被 import 时__name__是模块名。这个判断保证调试时直接跑被别的模块引用时不会自动执行。是写可复用 Python 模块的必备姿势。语法点 2\033[36ms06 \033[0m\033是ESC 转义字符[36m表示青色文字[0m表示重置样式。这是 ANSI 终端控制码俗称命令行里的 HTML 标签。效果提示符显示为青色视觉上区分用户输入和 Agent 输出。语法点 3except (EOFError, KeyboardInterrupt):EOFError用户按CtrlD在 Unix 系统里表示输入结束。KeyboardInterrupt用户按CtrlC中断信号。两个都捕获是因为终端交互里这两种退出意图都要优雅处理——直接break退出而不是抛异常栈。语法点 4query.strip().lower() in (q, exit, )链式调用先strip()去首尾空白再lower()转小写。in (q, exit, )判断是否在元组里。用元组而不是列表是因为元组不可变、更轻、更适合做常量枚举。一个 Python 惯用法用户可能输入多种退出方式→ 用in一次性判断。五、几个容易踩的坑坑 1子 Agent 的 System Prompt 写错了我一开始把run_subagent里的 SYSTEM 写成了SYSTEM主 Agent 的结果子 Agent 也在那派发任务越权递归。正确做法主 System 管规划子 System 管执行两句提示词职责清晰。坑 2tool_call_id对不上messages里assistant的tool_calls和后续role: tool消息的tool_call_id必须一一对应否则 API 直接 400。这是 OpenAI 兼容接口最坑人的地方。坑 3上下文压缩没做跑几步就爆代码里用[:50000]只是暴力截断生产环境至少要保留最近 N 轮完整对话更早的做摘要压缩工具结果超长时只存摘要 引用地址坑 4子 Agent 时间无限长for _ in range(30)只是防死循环真实场景还得加总时长超时工具调用次数上限Token 预算否则一个跑飞的子 Agent 能烧掉你一个月咖啡钱。坑 5可变默认参数上面agent_loop(messages: list)没设默认值是刻意的。如果你写def agent_loop(messages[])那个空列表会在所有调用之间共享——第二次调用时里面还残留第一次的数据。Python 里正确的写法是messagesNone然后在函数体内if messages is None: messages []。坑 6异常吞太狠except Exception as e很方便但会连KeyboardInterruptCtrlC之外的所有错误都吞掉包括你本应看到的 bug。建议在关键路径上记日志别让 Agent 静默失败。六、进阶从能跑到能上生产上面的代码能跑通 Sub-Agent 的核心机制但距离生产 Harness 还有一段路。这 4 个方向建议逐步补齐6.1 Observability能回放才敢上生产每次 Agent 执行打一个 Trace{ trace_id: ..., steps: [ {type: llm_call, tokens: 1200, latency: 800}, {type: tool_call, name: bash, args: {...}}, {type: subagent_spawn, prompt: ...} ] }没有 Trace 的 Agent等于闭着眼开车。6.2 Hooks可插拔的拦截点在 Loop 里留钩子让你能在不改进主流程的情况下插入工具调用前权限校验LLM 返回后输出过滤Agent 结束前质量评估6.3 Policy Safety权限与配额每个子 Agent 只能访问指定目录危险命令黑名单代码里的dangerous列表只是最简版资源配额Token / 时间 / 调用次数6.4 Memory跨会话的持久化history只存在内存里进程一挂就没了。生产环境要短期记忆当前会话Redis长期记忆向量库 结构化存储七、总结Harness Engineer 是个新工种回到开头那句话Harness Engineering 用工程化手段让 AI Agent 可靠、持续地完成任务。它不是 Prompt Engineering 的升级版而是软件工程 系统设计 LLM 能力的交叉地带。未来 2-3 年会出现这几类岗位AI 应用开发工程师—— 做 Agentic RAG 应用AI Agent 开发工程师—— 做 Harness、FDEForward Deployed EngineerAgent 架构师—— 设计多 Agent 协作体系6 大基础模块是操作系统内核Sub-Agents 是多进程管理Hooks/Safety 是权限系统。你现在缺的不是更牛的模型是一套能让模型乖乖干活的骨架。
