Agent Loop 是心跳:think -> act -> observe -> think。
它能让智能体动起来,但一上生产就会暴露问题:
Graph Engineer 的思路是:别把流程藏在 while 里,把流程显式画成图。 节点是动作,边是控制流,状态是共享黑板。
下面代码只做一件事: 用图驱动 Agent Loop,并带上预算、检查点和可观测。
import json, time
def checkpoint(state):
with open("state.json", "w") as f:
json.dump(state, f, ensure_ascii=False)
def run_graph(state, nodes, edges, start="planner", max_steps=20):
node = start
for _ in range(max_steps):
t = time.time()
state = nodes[node](state)
state["trace"].append({"node": node, "ms": int((time.time() - t) * 1000)})
checkpoint(state)
node = edges[node](state)
if node == "end":
return state
raise RuntimeError("budget exceeded")
def planner(s):
return {**s, "outline": ["引子", "原理", "代码", "总结"], "i": 0}
def writer(s):
i = s["i"]
return {**s, "draft": s["draft"] + f"\n## {s['outline'][i]}\n...", "i": i + 1}
def reviewer(s):
return {**s, "pass": s["i"] >= len(s["outline"])}
NODES = {"planner": planner, "writer": writer, "reviewer": reviewer}
EDGES = {
"planner": lambda s: "writer",
"writer": lambda s: "reviewer",
"reviewer": lambda s: "end" if s["pass"] else "writer",
}
state = {"topic": "Agent Loop + Graph Engineer", "draft": "", "trace": []}
result = run_graph(state, NODES, EDGES)
print(result["draft"])这段代码把 Agent Loop 工程化了:
draft、outline、i、pass、trace 都在状态里reviewer 决定回退还是结束max_steps 防止无限循环state.jsontrace 记录节点和耗时少量代码验证骨架,生产环境还要加:
Agent Loop 是执行引擎,Graph Engineer 是工程骨架。 少量代码就能把循环变成图:节点做事,边做路由,状态做真相。
但真正决定能不能上生产的,是预算、检查点、可观测和安全边界。 图让智能体可解释,工程让它可运维。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。