AI Agent 应用实战(3):构建单 Agent 任务执行循环
上一篇把单次工具调用收进白名单和参数校验。本篇继续向前一步:让 Agent 在“观察—选择动作—执行—再观察”之间循环,同时用状态机、预算和重复检测保证它一定停下来。
一、痛点:从“能演示”走向“可托管”
循环不是一个无限 while,而是受控状态机。运行状态至少包含目标、事实、动作历史、剩余步数和终止原因;每轮只能产生 tool、final 或 abort 三类决定。执行器先检查预算和重复动作,再调用工具并追加观察。模型不能改写历史,也不能宣布已经执行了一个实际未执行的动作。
真正可迁移的做法,是先写任务契约再选模型:输入从哪里来,成功产物是什么,哪些动作禁止自动执行,最多允许多少步、多少时间和多少费用,失败后由谁接管。契约一旦写进代码和测试,模型升级只是实现替换,不会悄悄改变业务责任边界。
二、原理:把概率判断放进确定性边界
步数、墙钟时间和费用是三套独立预算,任意一项耗尽都要停止。动作去重应基于规范化参数后的哈希,否则只改字段顺序就能绕过。终态需要区分 completed、needs_human、budget_exhausted 与 failed,调用方才能采取正确后续动作,而不是把所有非成功都包装成模糊回复。
模型输出、工具数据和记忆内容都按不可信输入处理。每个边界都执行校验、授权、裁剪和审计;所有状态变化都有明确原因。这样做看似增加一些代码,却把“偶尔答错”拆成可定位的规划错误、参数错误、工具错误或策略错误,团队才能针对性改进。
三、实现:用独立程序跑通最小闭环
示例把“查询订单后计算是否满足补偿条件”拆成确定动作序列。计划器在这里用固定规则模拟,便于观察循环机制;接入模型时只替换 decide,状态、工具、预算和终态判断无需改变。
from dataclasses import dataclass, field
@dataclass
class State:
facts: dict[str, int] = field(default_factory=dict)
history: list[str] = field(default_factory=list)
remaining: int = 4
status: str = "running"
def decide(state: State) -> str:
if "order_age" not in state.facts:
return "lookup_order"
if "max_age" not in state.facts:
return "read_policy"
return "finish"
def execute(action: str, state: State) -> str:
if action == "lookup_order":
state.facts["order_age"] = 5
return "order_age=5"
if action == "read_policy":
state.facts["max_age"] = 7
return "max_age=7"
eligible = state.facts["order_age"] <= state.facts["max_age"]
state.status = "completed"
return f"eligible={str(eligible).lower()}"
state = State()
while state.status == "running" and state.remaining:
action = decide(state)
observation = execute(action, state)
state.history.append(action)
state.remaining -= 1
print(f"step={len(state.history)} action={action} observation={observation}")
if state.status == "running":
state.status = "budget_exhausted"
print(f"status={state.status} remaining={state.remaining}")
运行输出:
step=1 action=lookup_order observation=order_age=5
step=2 action=read_policy observation=max_age=7
step=3 action=finish observation=eligible=true
status=completed remaining=1
可恢复性比循环写法更重要。每完成一步就持久化检查点,并为写操作保存幂等键。进程在工具成功后、状态落盘前崩溃,是最危险的窗口;恢复时必须先查询幂等记录,不能盲目重放。观察值要有大小上限和来源标签,避免一次网页抓取占满上下文。
第二个程序补上生产中最容易遗漏的控制点。它与前一个示例完全独立,可单独保存运行,不依赖本系列其他文件;示例数据是内存假实现,因此不会访问真实账户或产生外部副作用。
import json
import hashlib
def fingerprint(name: str, arguments: dict[str, str]) -> str:
payload = json.dumps(arguments, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
digest = hashlib.sha256(f"{name}:{payload}".encode()).hexdigest()
return digest
seen: set[str] = set()
actions = [
("search_order", {"id": "A-7"}),
("search_order", {"id": "A-7"}),
("read_policy", {"version": "2026-01"}),
]
for name, arguments in actions:
key = fingerprint(name, arguments)
normalized = json.dumps(arguments, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
if key in seen:
print("blocked=duplicate_action")
continue
seen.add(key)
print(f"accepted={name}:{normalized}")
print(f"unique_actions={len(seen)}")
运行输出:
accepted=search_order:{"id":"A-7"}
blocked=duplicate_action
accepted=read_policy:{"version":"2026-01"}
unique_actions=2
落地时应把示例中的内存状态替换为事务型存储,把打印日志替换为结构化事件,但不要改变契约。事件至少包含 run_id、步骤、输入摘要、策略结果、耗时、错误类别和版本;密钥、完整个人信息及未经脱敏的提示不得进入日志。
四、踩坑:失败路径决定系统上限
循环最常见的故障是振荡:Agent 在两个工具间来回切换,或者用同一查询反复碰运气。另一个问题是“伪完成”,模型生成答案却缺少任务要求的证据。执行器应设置重复上限,final 也要通过完成条件校验,例如订单与政策两份事实都存在;否则转人工,而不是继续消耗预算。
还要防止把确定逻辑模型化。金额计算、权限判断、枚举路由和状态转移应使用普通代码;模型适合处理分类、抽取、歧义消解与证据综合。每减少一个无必要的模型决策点,就减少一处延迟、成本和不可复现性,同时让测试更容易覆盖。
五、验证:用轨迹和门槛验收
回归用完整轨迹验收,而不只看最后一句话。每个案例检查动作是否合法、参数是否稳定、观察是否被正确利用、是否在预算内进入预期终态。加入空结果、超时、重复建议和进程恢复用例。下一篇会把长轨迹中的信息分层保存,解决上下文越跑越长的问题。
发布顺序固定为离线回放、影子流量、小比例灰度和有限自治。每阶段先定义通过线和回滚条件,再看结果;不能在看到分数后移动门槛。关键样本要保存初始状态、每步动作、观察、最终产物与终止原因,模型、提示和工具契约版本也必须一起记录。
循环的状态将直接成为记忆管理的输入:哪些是当前步骤临时信息,哪些是会话事实,哪些才值得跨会话保存。
参考来源
👍 觉得有用就点个 赞 + 收藏,方便回头查阅;有疑问直接在评论区留言,我看到都会回。
🚀 本文属于 《AI Agent 应用实战》 系列,持续更新,关注不迷路。
📌 文章里的代码都能直接跑。想要可直接 clone 的完整工程 + 配套部署脚本 / 踩坑清单?评论一声或发邮件到 cj2664@qq.com,我免费发你。
如果你正好在做类似系统、或有工程化难题想找人做,也欢迎邮件聊一句——我按实际情况评估,能落地的就接单或出方案。评论和邮件都能直接找到我,不用跳别的平台。
更多推荐



所有评论(0)