Harness 是地基层,Loop 和 Graph 是控制流层的两种形态—— 一篇写给工程师的祛魅文章
如果你最近在看 Agent 相关的技术讨论,大概率会撞见三个听起来很像、又说不清区别的词:Agent Harness Engineering、Loop Engineering、Graph Engineering。它们的尴尬之处在于:三个词都在描述“模型之外的那部分工程”,但切的维度完全不同。于是你会看到“Graph 是 Loop 的升级版”“有了 Harness 就不需要 Graph”这类很奇怪的争论。
这些说法都不对,因为它们把同一层的东西和不同层的东西混在一起比了。Loop 和 Graph 互为替代:在同一个任务上二选一;而 Harness 和它们不是替代关系:无论选 Loop 还是 Graph,都需要 Harness。如果这句话已经懂了,后面可以直接看代码部分;如果还没懂,我们从最原始的地方开始。
01、先说清楚:这三个词为什么会打架
先给一个结论,后面的篇幅都是在论证它。
Harness 是地基层,Loop 和 Graph 是控制流层的两种形态。Loop 和 Graph 互为替代(在同一个任务上二选一),而 Harness 和它们不是替代关系:无论你选 Loop 还是 Graph,你都需要 Harness。
02、一个 while 循环能走多远
2.1 定义
Loop Engineering 是指:把 Agent 的控制流建模为“单一的、无固定终点的迭代循环”,通过模型自身的决策来驱动每一轮的行为选择和终止判断。它的骨架朴素到令人不安:
while 未完成:
模型看当前上下文 → 决定做什么
执行 → 拿到结果 → 塞回上下文
没有预定义的步骤,没有流程图,没有状态机。模型自己就是调度器。这就是 ReAct 范式的本质,也是 Claude Code、Codex CLI、Cursor Agent 这类工具在底层跑的东西。很多人第一次看到这些产品的架构图时会失望:就这?一个 while 循环?对,就这。但“就这”背后有大量工程。
2.2 最小可运行的 Loop
不用任何框架,裸 API:
...run_loop.py
import json
from openai import OpenAI
client = OpenAI(
api_key="sk-xxxx",
base_url="https://api.deepseek.com"
)
TOOLS = [{
"type": "function",
"function": {
"name": "read_file",
"description": "读取一个文件的内容",
"parameters": {
"type": "object",
"properties": {"path": {"type": "string"}},
"required": ["path"]
}
}
}, {
"type": "function",
"function": {
"name": "write_file",
"description": "写入内容到文件",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"}
},
"required": ["path", "content"]
}
}
}]
def dispatch(name, args):
if name == "read_file":
with open(args["path"], "r", encoding="utf-8") as f:
return f.read()
if name == "write_file":
with open(args["path"], "w", encoding="utf-8") as f:
f.write(args["content"])
return f"已写入 {args['path']}"
return f"未知工具: {name}"
def run_loop(task: str, max_turns: int = 30):
messages = [
{"role": "system", "content": "你是一个代码助手,通过工具完成任务。完成后直接回复结论,不要再调用工具。"},
{"role": "user", "content": task}
]
for turn in range(max_turns):
resp = client.chat.completions.create(
model="deepseek-v4-flash",
messages=messages,
tools=TOOLS
)
msg = resp.choices[0].message
messages.append(msg.model_dump(exclude_none=True))
# 模型不再调工具 = 它认为任务结束了
if not msg.tool_calls:
return msg.content
for call in msg.tool_calls:
args = json.loads(call.function.arguments)
try:
result = dispatch(call.function.name, args)
except Exception as e:
result = f"执行失败: {type(e).__name__}: {e}"
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": str(result)[:8000] # 截断,防止污染上下文
})
return "达到最大轮次上限,任务未完成"
if __name__ == "__main__":
result = run_loop("读取 test.txt,然后告诉我里面写了什么")
print(result)
...run_loop.py 执行结果
#我们在目录下创建一个 test.txt 文件,写入 Hello字符。
# 输出信息
文件读取完成!**test.txt** 里面写的内容很简单,就是:**Hello**。
八十行不到,这就是一个能干活的 Agent。
2.3 Loop 的核心特征
控制权在模型手里:下一步做什么、什么时候停,都是模型在 token 层面决定的,你写的代码只是个执行器。状态就是消息数组,没有额外的状态对象,messages 本身承载了全部历史,这是 Loop 最优雅的地方,也是它最脆弱的地方。路径是涌现的,不是设计的。同样的任务跑两次,模型可能走完全不同的步骤,这在探索性任务里是优点,在需要审计的业务流程里是灾难。
2.4 Loop 的真实痛点
写出上面那段代码只要一小时,让它在生产环境跑一个月不出事,要一个季度。因为你会依次撞上:
上下文爆炸
一个复杂任务跑三十轮,每轮工具返回几千 token,上下文很快撑爆,需要压缩、摘要、分页、外置存储。
无限循环
模型会陷入“读文件 → 发现不对 → 再读同一个文件”的死循环,max_turns 只是止损,不是解法。
错误级联
第 5 轮一个工具返回了脏数据,模型信了,后面 25 轮全部建立在错误前提上。
不可复现
出了问题没法说“从第 12 步重跑”,因为根本没有“第 12 步”这个概念,只有一段线性的消息流。
权限失控
模型某一轮突然决定 write_file(“/etc/passwd”, ...),你的循环会老老实实执行。
这些痛点里,有的靠改造循环本身解决不了,于是催生了另外两个词。
03、把控制权夺回来
3.1 定义
Graph Engineering 是指:把 Agent 的执行过程显式建模为有向图:节点是确定性的处理单元,边是转移条件,状态在节点间显式流动。如果说 Loop 是“让模型自由发挥”,Graph 就是“我来画好路线,模型只在指定路口做选择”。关键区别在于控制权的归属:
| 范式 | 谁决定下一步 | 谁决定何时停止 |
|---|---|---|
| Loop | 模型 | 模型 |
| Graph | 图的边(代码) | 图的终止节点(代码) |
模型在 Graph 里依然重要,但它从“调度器”降级为“节点内部的一个能力”。
3.2 手写一个最小的 Graph 引擎
...graph_engine.py
from __future__ import annotations
import json
import os
import sys
import time
from dataclasses import dataclass, field
from typing import Callable, TypedDict
from dotenv import load_dotenv
load_dotenv()
MODEL = "deepseek-v4-flash"
MAX_REPLANS = 2 # 重规划次数上限:写死在代码里,不交给模型自觉
MAX_STEPS = 50 # 图的总步数熔断
# ============================================================
# 1. 状态定义
# ============================================================
class State(TypedDict, total=False):
task: str # 用户输入
plan: list[str] # 拆解出的步骤
cursor: int # 当前执行到第几步
results: list[str] # 每步的产出
review: str # 审查结论
replans: int # 已重规划次数
final: str # 最终回答
trace: list[str] # 执行轨迹(可观测性)
# ============================================================
# 2. 图引擎
# ============================================================
END = "END"
@dataclass
class Graph:
entry: str = ""
nodes: dict[str, Callable[[State], State]] = field(default_factory=dict)
edges: dict[str, Callable[[State], str]] = field(default_factory=dict)
def node(self, name: str):
"""装饰器:注册一个节点"""
def deco(fn):
self.nodes[name] = fn
return fn
return deco
def edge(self, src: str, router: Callable[[State], str]):
"""注册一条出边(路由函数)"""
self.edges[src] = router
def validate(self) -> None:
"""启动前静态检查:比运行时炸掉便宜得多"""
assert self.entry in self.nodes, f"入口节点不存在: {self.entry}"
for name in self.nodes:
assert name in self.edges, f"节点 {name} 没有出边"
def run(self, state: State) -> State:
self.validate()
state.setdefault("trace", [])
cur = self.entry
for step in range(MAX_STEPS):
if cur == END:
break
t0 = time.time()
state = self.nodes[cur](state)
nxt = self.edges[cur](state)
state["trace"].append(
f"#{step:02d} {cur:>10s} --({int((time.time()-t0)*1000)}ms)--> {nxt}"
)
cur = nxt
else:
state["trace"].append(f"!! 达到步数上限 {MAX_STEPS},强制终止")
return state
# 断点续跑:状态是显式对象,所以这两个方法几乎是免费的
def save(self, state: State, path: str) -> None:
with open(path, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
@staticmethod
def load(path: str) -> State:
with open(path, encoding="utf-8") as f:
return json.load(f)
# ============================================================
# 3. 模型调用(带 mock 兜底)
# ============================================================
MOCK = "--mock" in sys.argv or not os.getenv("DEEPSEEK_API_KEY")
_client = None
def _get_client():
global _client
if _client is None:
from openai import OpenAI
_client = OpenAI(
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url="https://api.deepseek.com/v1",
)
return _client
def llm(prompt: str, system: str = "你是一个严谨的工程助手,回答简洁不啰嗦。") -> str:
if MOCK:
return _mock_llm(prompt)
try:
resp = _get_client().chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
temperature=0.3,
)
return (resp.choices[0].message.content or "").strip()
except Exception as e:
# 节点内的模型失败不应该炸掉整张图,降级为可路由的文本
return f"[LLM_ERROR] {type(e).__name__}: {e}"
_MOCK_CALLS = {"review": 0}
def _mock_llm(prompt: str) -> str:
"""离线桩:让图结构可以在没有 key 的情况下完整跑通,包括 FAIL 分支"""
if "拆成" in prompt:
return "1. 明确输入输出格式\n2. 设计核心数据结构\n3. 给出关键代码骨架"
if "只回答 PASS 或 FAIL" in prompt:
_MOCK_CALLS["review"] += 1
if _MOCK_CALLS["review"] == 1:
return "FAIL - 缺少错误处理的说明,建议补充"
return "PASS - 步骤完整,结论可用"
if "整理成一份" in prompt:
return "【最终方案】\n(mock 模式产出,接入真实 key 后由模型生成)"
return f"(mock 结果)已处理:{prompt.splitlines()[-1][:40]}"
# ============================================================
# 4. 节点实现
# ============================================================
g = Graph(entry="plan")
@g.node("plan")
def plan_node(state: State) -> State:
# 重规划时把上一轮的审查意见带上,否则模型会原地重复同样的错误
feedback = ""
if state.get("review"):
feedback = (
f"\n\n上一轮方案的问题:{state['review']}\n"
f"上一轮产出摘要:{' | '.join(r[:60] for r in state.get('results', []))}\n"
"请针对上述问题重新拆解。"
)
raw = llm(
f"把下面的任务拆成 3-6 个可独立执行的步骤,每行一个,"
f"只输出步骤本身,不要任何额外解释:\n{state['task']}{feedback}"
)
plan = [ln.strip().lstrip("0123456789.、) ") for ln in raw.splitlines() if ln.strip()]
state["plan"] = plan[:6] or [state["task"]] # 兜底:拆解失败就当单步任务
state["cursor"] = 0
state["results"] = [] # 重规划时清空产出,但 feedback 已保留信息
return state
@g.node("execute")
def execute_node(state: State) -> State:
step = state["plan"][state["cursor"]]
# 只带最近 3 步上下文 —— 上下文预算由图结构约束,不靠模型自律
ctx = "\n".join(state["results"][-3:]) or "(这是第一步)"
out = llm(
f"总任务:{state['task']}\n\n已完成的部分:\n{ctx}\n\n"
f"现在只执行这一步,不要越界:{step}"
)
state["results"].append(f"[{step}]\n{out}")
state["cursor"] += 1
return state
@g.node("review")
def review_node(state: State) -> State:
state["review"] = llm(
f"任务:{state['task']}\n\n执行结果:\n" + "\n\n".join(state["results"]) +
"\n\n判断任务是否已完整达成。只回答 PASS 或 FAIL,后接一行理由。"
)
return state
@g.node("finalize")
def finalize_node(state: State) -> State:
state["final"] = llm(
"把下面的执行结果整理成一份给用户看的、结构清晰的最终回答:\n\n"
+ "\n\n".join(state["results"])
)
return state
# ============================================================
# 5. 边:全部控制权在这里
# ============================================================
def route_execute(s: State) -> str:
return "execute" if s["cursor"] < len(s["plan"]) else "review"
def route_review(s: State) -> str:
verdict = s["review"].upper()
if verdict.startswith("PASS") or "[LLM_ERROR]" in verdict:
return "finalize"
if s.get("replans", 0) < MAX_REPLANS:
s["replans"] = s.get("replans", 0) + 1 # 关键:自增在这里,否则死循环
return "plan"
return "finalize" # 用尽重试也要给用户一个结果
g.edge("plan", lambda s: "execute")
g.edge("execute", route_execute)
g.edge("review", route_review)
g.edge("finalize", lambda s: END)
我们实现4个节点(plan → execute → review → finalize):下面我们调用Graph引擎。
# ============================================================
# 6. 入口
# ============================================================
def main() -> None:
args = [a for a in sys.argv[1:] if not a.startswith("--")]
task = args[0] if args else "设计一个能统计 Nginx 访问日志 Top10 IP 的小工具"
if MOCK:
print(">>> mock 模式(未检测到 DEEPSEEK_API_KEY),仅验证图结构\n")
final_state = g.run({"task": task})
print("=" * 60)
print("执行轨迹")
print("=" * 60)
for line in final_state["trace"]:
print(line)
print("\n" + "=" * 60)
print(f"最终回答 (重规划 {final_state.get('replans', 0)} 次)")
print("=" * 60)
print(final_state.get("final", "(无产出)"))
g.save(final_state, "graph_state.json")
print("\n状态已存至 graph_state.json —— 可用 Graph.load() 断点续跑")
if __name__ == "__main__":
main()
#输出节选
============================================================
执行轨迹
============================================================
#00 plan --(3096ms)--> execute
#01 execute --(2172ms)--> execute
#02 execute --(3365ms)--> execute
#03 execute --(2836ms)--> execute
#04 execute --(2225ms)--> execute
#05 execute --(1324ms)--> execute
#06 execute --(961ms)--> review
#07 review --(2316ms)--> finalize
#08 finalize --(3987ms)--> END
============================================================
最终回答 (重规划 0 次)
============================================================
以下是根据您提供的代码片段整合后的完整脚本及解释,用于统计 Nginx 访问日志中访问次数最多的前 10 个 IP 地址:
```python
import re
# 1. 读取并解析日志行,提取客户端 IP
def parse_log_line(line: str) -> str | None:
# 匹配行首的 IPv4 地址
LOG_PATTERN = r'^(\d+\.\d+\.\d+\.\d+)'
match = re.match(LOG_PATTERN, line)
return match.group(1) if match else None
# 2. 统计每个 IP 出现次数
以下省略......
注意 review 那条边:重规划的次数是代码里的硬约束,不是模型的自觉。这就是 Graph 相对 Loop 最本质的收益。
3.3 Graph 带来了什么
Graph 可预测执行路径的全集是有限的、可枚举的,可以画在白板上给产品经理看。可断点续跑:状态是显式对象,序列化存下来,进程崩了可以从任意节点恢复;Loop 想做到这一点得把整个消息数组存下来,而且恢复后模型未必还认得当时的语境。可并行:图里没有依赖关系的节点可以并发跑,线性的 Loop 天然做不到。可局部替换:某个节点的模型效果不好,单独换掉、单独测试、单独降级到规则实现;Loop 里没有“某个节点”这个概念。天然的上下文隔离:每个节点只看它需要的状态字段,不是整个历史,上下文爆炸问题被结构性地缓解了。
3.4 Graph 的代价
灵活性坍塌:图画好那一刻,Agent 的能力上限就定死了,用户提了个没设计过的需求,图跑不通。设计成本高:四节点的图已经要写几十行路由逻辑,真实业务里二十个节点、带并行分支和错误分支,那是一张需要专门维护的架构图。过度工程的诱惑:很多团队一上来就画图,结果发现 80% 的路径从来没被触发过,他们本来用一个 Loop 就够了。探索性任务的死穴:“帮我调查一下这个 bug”,你没法给“调查”画一张图,因为调查的路径取决于调查过程中发现了什么。
04、不是控制流,是运行时
到这里,很多人会以为下一步是“Harness = Loop 和 Graph 的某种混合”。不是。Harness 根本不在这一层。
4.1 定义
Agent Harness Engineering 是指:构建模型与真实世界之间的那层运行时基础设施:工具接口、上下文管理、权限边界、可观测性、容错恢复、成本控制。它决定的不是“Agent 走哪条路”,而是“Agent 在什么样的环境里跑”。Harness 这个词来自机械领域:马的挽具。挽具不决定马往哪走,它决定马的力量怎么传导到车上、怎么被约束、怎么在失控时被拉住。一个更工程化的类比:如果模型是 CPU,Loop/Graph 是你写的程序,那 Harness 就是操作系统。
4.2 Harness 包含什么
这不是一个抽象概念,它有非常具体的组成:
工具层(Tool Surface)
工具粒度设计、工具描述的写法、返回值的格式化与截断、工具的幂等性与可重试性。
上下文层(Context Management)
历史超阈值时的压缩摘要、把大块内容卸载到外部存储只留句柄、需要时再检索回来——这一层现在被单独叫做 Context Engineering,是 Harness 的核心子集。
安全层(Guardrails)
工具调用白名单与参数校验、危险操作人工确认、提示注入防御(工具返回的内容是数据不是指令)、沙箱与文件系统边界。
可观测层(Observability)
每次模型调用的完整 trace、token 消耗、延迟、工具成功率、失败案例的可回放性。
容错层(Recovery)
工具异常时返回结构化错误让模型自纠、超时限流与重试退避、预算熔断(任务烧了 5 块钱就停)。
4.3 Harness 的代码形态
Harness 通常表现为装饰器、中间件和包装层,而不是主流程:
...harness.py
import time, json, logging
from functools import wraps
log = logging.getLogger("harness")
class Budget:
def __init__(self, max_tokens=200_000, max_calls=50):
self.max_tokens, self.max_calls = max_tokens, max_calls
self.tokens, self.calls = 0, 0
def check(self):
if self.tokens > self.max_tokens:
raise BudgetExceeded(f"token 超预算: {self.tokens}")
if self.calls > self.max_calls:
raise BudgetExceeded(f"调用次数超限: {self.calls}")
class BudgetExceeded(Exception): pass
DANGEROUS = {"write_file", "run_shell", "delete"}
def harness(budget: Budget, confirm=None):
"""把一个裸的工具函数包装成生产可用的工具"""
def deco(fn):
@wraps(fn)
def wrapper(**kwargs):
name = fn.__name__
t0 = time.time()
# 1. 安全:危险操作需要确认
if name in DANGEROUS and confirm and not confirm(name, kwargs):
return {"ok": False, "error": "用户拒绝了该操作"}
# 2. 预算:先检查
budget.calls += 1
try:
budget.check()
except BudgetExceeded as e:
return {"ok": False, "error": str(e), "fatal": True}
# 3. 容错:异常转为结构化返回,而不是抛出
try:
raw = fn(**kwargs)
except Exception as e:
log.exception("tool failed: %s", name)
return {"ok": False,
"error": f"{type(e).__name__}: {e}",
"hint": "请检查参数或换一种方式"}
# 4. 上下文:超长结果卸载到文件
text = raw if isinstance(raw, str) else json.dumps(raw, ensure_ascii=False)
if len(text) > 6000:
path = f"/tmp/tool_out_{int(time.time()*1000)}.txt"
with open(path, "w", encoding="utf-8") as f:
f.write(text)
text = (text[:2000] +
f"\n\n...(结果过长已截断,完整内容见 {path},"
f"共 {len(text)} 字符,可用 read_file 分段读取)")
# 5. 可观测:结构化日志
log.info(json.dumps({
"tool": name, "ms": int((time.time()-t0)*1000),
"in_bytes": len(str(kwargs)), "out_bytes": len(text)
}, ensure_ascii=False))
return {"ok": True, "data": text}
return wrapper
return deco
budget = Budget(max_tokens=150_000, max_calls=40)
用起来是这样:
budget = Budget(max_tokens=150_000, max_calls=40)
@harness(budget, confirm=lambda n, a: input(f"允许 {n}({a})? [y/N] ") == "y")
def write_file(path: str, content: str):
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return f"已写入 {path},{len(content)} 字符"
if __name__ == "__main__":
result = write_file(
path="hello.txt",
content="你好,Agent Harness"
)
print(result)
执行输出:
允许 write_file({'path': 'hello.txt', 'content': '你好,Agent Harness'})? [y/N] y
{'ok': True, 'data': '已写入 hello.txt,16 字符'}
注意:这段代码里完全没有出现 while 循环,也没有出现节点和边。它对上层控制流是完全无感的:可以套在 Loop 上,套在 Graph 的节点里也可以。这就是为什么说 Harness 和 Loop/Graph 不在一个维度上。
4.4 一个被低估的事实
业界有个逐渐形成的共识:同一个模型,配不同的 Harness,表现差距可以大过换模型。这不难理解:模型能力是给定的,但 Harness 决定了模型能看到多少有效信息、动作能不能可靠地作用于世界、犯错后能不能恢复、作死时会不会真的把系统搞挂。Claude Code 这类产品的护城河,大部分不在模型侧,而在 Harness 侧,这也是为什么它们的控制流看起来简单到没有技术含量:复杂度被转移到了 Harness 层。
05、三者的正确关系图
控制流层 Loop Engineering 或 Graph Engineering
← 二选一 / 混用,谁决定下一步
运行时层 Agent Harness Engineering
← 恒定需要,工具 / 上下文 / 安全 / 观测 / 容错
模型层 deepseek-v4-flash / ...
三句话总结:
1、Loop 和 Graph 是替代关系:针对同一个任务,你在两者中选一个(或者混用)。
2、Harness 和它们是正交关系:不管上面选什么,下面都需要 Harness。
3、没有 Harness 的 Loop 是玩具;没有 Harness 的 Graph 是脆的流水线。
5.1 三者互补的典型形态:Graph 套 Loop
生产系统里最实用的架构,往往不是纯 Loop 也不是纯 Graph,而是外层用 Graph 保证流程可控,某些节点内部跑 Loop 保证灵活性:
...graph_wraps_loop.py
@g.node("investigate")
def investigate_node(state: State) -> State:
"""这个节点内部是一个受限的 Loop——因为'调查'无法画成图"""
state["findings"] = run_loop(
task=f"调查以下问题并给出根因:{state['issue']}",
max_turns=15 # Graph 层给 Loop 套上轮次上限
)
return state
@g.node("apply_fix")
def apply_fix_node(state: State) -> State:
"""这个节点是确定性的——修复动作必须可审计,不能让模型自由发挥"""
patch = llm(f"根据根因生成修复补丁:\n{state['findings']}")
state["patch"] = patch
return state
g.edge("investigate", lambda s: "apply_fix" if s.get("findings") else "escalate")
g.edge("apply_fix", lambda s: "human_review") # 强制人工审核,写死在边上
在“过程不可预测但结果可验证”的地方用 Loop,在“过程必须可审计”的地方用 Graph
调查、探索、检索、debug用 Loop。审批、支付、写库、发布用 Graph。而 Harness 在两边都跑着,不管你在哪一层。
06、怎么选:一张决策表
| 判断维度 | 倾向 Loop | 倾向 Graph |
|---|---|---|
| 路径可预知吗 | 不可预知 | 可枚举 |
| 需要审计/合规吗 | 不需要 | 需要 |
| 步骤数量 | 高度可变(3~50) | 相对固定 |
| 失败代价 | 低,可重试 | 高,不可逆 |
| 需要并行吗 | 不需要 | 需要 |
| 迭代速度要求 | 快速试错 | 稳定交付 |
| 典型场景 | 代码助手、研究、客服排障 | 订单处理、内容审核、财务对账 |
还有两条经验法则:
法则一:先写 Loop,撞墙了再改 Graph
先用 Loop 跑通,收集真实的失败案例。你会发现失败集中在少数几个环节,把那几个环节固化成图,其余留给 Loop。反过来做(先设计完美的图)几乎必然过度工程。
法则二:Harness 要在第一天就开始建
Loop 还是 Graph 可以之后改,但如果第一天没有 trace、没有预算控制、没有工具错误的结构化返回,你后面每一次 debug 都是盲人摸象。这一层的技术债利息最高。
07、放在 AI 工程演进史里看
这三个概念的出现,对应的是一条清晰的重心迁移路线:
第一阶段 Prompt Engineering(2022–2023)
重心在“怎么把话说好”。模型能力弱,提示词的微小改动带来巨大差异,工程量集中在字符串里。
第二阶段 Context Engineering(2024–2025)
重心从“说什么”转向“给它看什么”。RAG、压缩、卸载、注意力预算成为核心问题,模型的表现是上下文质量的函数。
第三阶段 Harness / 控制流工程(2025–)
重心从“给它看什么”转向“让它在什么环境里、按什么结构行动”。瓶颈变成了多轮行动的可靠性:不失控、不失忆、不越权、可恢复、可观测。
这条线的本质是:工程的重心从模型内部,一路外移到模型周围。
这对工程师是好消息:
1、能力差异重新回到工程手里。大家用同样的模型,做出来的产品体验天差地别:差在 Harness 和控制流设计上,这是可以靠工程功力拉开的差距。
2、传统软件工程的经验重新值钱。状态机、幂等、熔断、超时重试、可观测性、权限模型,不是 AI 概念,是三十年的分布式系统经验;你过去写业务系统的所有直觉,在这里都还有效。
3、“框架”的价值被重新定价。LangGraph 之类提供的主要是 Graph 层抽象,但 Harness 层高度依赖具体业务,很难被框架完全封装,这也是为什么越来越多团队选择裸 API + 自建 Harness。
写在最后
回到开头那个混乱:有人说“Graph 是 Loop 的升级版”,错!它们是两种控制哲学,各有适用域,Loop 在探索性任务上不可替代。有人说“有了 Harness 就不需要 Graph”,错!它们不在一层,Harness 解决不了流程可审计的问题。有人说“这三个都是营销包装”,部分对!词是新的,但词背后的问题是真的、旧的、硬的:控制权归属、状态管理、故障恢复、边界约束。
别急着选词。先问自己三个问题:这个任务的路径可以枚举吗?(决定 Loop 还是 Graph)它失败的时候,我能看见吗?(决定 Harness 的观测层)它作死的时候,我拦得住吗?(决定 Harness 的安全层)
第一个问题决定架构,后两个问题决定你能不能上生产。大部分团队死在后两个上。
END
我是写代码的中年人,长期专注于 AI 算法、智能体、以及 AI 工程化落地。
128