拓展阅读

模型提议与程序执行分离

用确定性模拟模型依次提出读取、修改、测试。执行器检查工具和作用域,记录幂等操作,最终状态依赖验证。

状态:自检已实际通过 · 环境:Python标准库 · 性质:合成/模拟教学实验
对应讲义 5.1 · 保存Python脚本

沿着对象与状态理解本章

把模型、宿主、工具画成不同泳道后,问“谁真正改变了文件”就有明确答案。跟踪每条返回消息,再判断需要继续调用模型、执行工具还是结束;不要把模型说完成当成环境已经完成。

下方图解的固定输入在“输入与适用条件”中声明。它用于解释本章机制,不与本章其他独立例子的数值暗中接续。

动态图解 · F22

把一个Agent任务拆成真正发生的几轮

同一句用户任务,为什么可能调用模型不止一次?

LLM生成下一步提议,宿主执行获准的动作,把结果再交给LLM。这是外层任务循环;每次模型调用内部还要经历逐Token生成。外层的一轮工具行动不等于内部生成一个Token,两种“循环”不能混用。

调用标识把某次工具请求与其结果连接的ID。
工具结果程序实际执行后返回的内容。
上下文本轮明确交给模型的信息,而非整个电脑状态。

输入与适用条件:构造的只读教学任务:文件config.txt含timeout=30。泳道展示职责边界,不声称调用了真实商业API;数据和动作顺序固定。

1 / 8 · 1. 用户任务:读config.txt中的timeout

第1步,共8步

窄屏可左右滑动图形,或点击“放大图形”。文字解释紧随图后。

把一个Agent任务拆成真正发生的几轮:1. 用户任务:读config.txt中的timeout用户宿主LLM文件工具用户任务:读config.txt中的timeout模型请求:任务+read_file工具定义模型输出:call_id=c1,path=config.txt宿主校验后读取文件工具结果:timeout=30下一轮输入:原任务+调用+结果模型回答:配置的timeout为30宿主显示回答;任务结束

模型尚未看到磁盘文件。先区分用户给的任务与环境真实内容。

未读取

默认静止。用“下一步”看完解释再继续;播放到最后一步自动停止。键盘聚焦播放器后,可用左右键逐步查看、空格播放/暂停。图中颜色辅助区分,含义同时用文字注明。

为什么需要

模型可以借助工具获得新证据,而不把一切都塞入参数。

怎样实现

按请求、校验、执行、结果回填、继续决策组织循环。

可能的代价与误解

忘记回填工具结果会让模型重复请求;把输出文本当执行证明会误报完成。

如何判断是否做对

对照每轮模型输入和工具日志,确认结果与call_id匹配。

怎样观察与验证

完整任务只有一次模拟写入,重复同一操作ID不重复写。未授权publish被拒绝。没有调用真实LLM,也不读写真实工程文件。

运行命令

python3 reading-notes/experiments/07_agent_loop.py

完整教学代码

"""确定性模拟模型与工具循环;不调用LLM、不读写实际工程文件。"""
from dataclasses import dataclass, field


@dataclass
class State:
    evidence: bool=False
    patched: bool=False
    verified: bool=False
    events: list[dict]=field(default_factory=list)


class Executor:
    def __init__(self, allowed: set[str]):
        self.allowed=allowed
        self.completed: dict[str,tuple[dict,dict]]={}
        self.write_count=0

    def execute(self, call: dict, state: State) -> dict:
        if set(call)!={"id","tool","args"}:
            raise ValueError("调用字段必须为 id/tool/args")
        if not isinstance(call["id"],str) or not call["id"]:
            raise ValueError("调用ID必须是非空字符串")
        if call["tool"] not in self.allowed:
            raise PermissionError("工具未授权")
        if call["args"] != {"scope":"toy-project"}:
            raise PermissionError("作用域越界")
        if call["id"] in self.completed:
            previous,result=self.completed[call["id"]]
            if previous != call:
                raise ValueError("同一幂等ID不能指向不同动作")
            return {**result,"replayed":True}
        name=call["tool"]
        if name=="read":
            state.evidence=True
            result={"status":"ok","evidence":"toy:T1"}
        elif name=="patch":
            if not state.evidence:
                raise RuntimeError("缺少前置证据")
            state.patched=True; self.write_count+=1
            result={"status":"ok","version":"toy:v2"}
        elif name=="test":
            state.verified=state.patched
            result={"status":"passed" if state.verified else "failed"}
        else:
            raise ValueError("未知教学工具")
        self.completed[call["id"]]=(dict(call),dict(result))
        return result


def fake_model(state: State) -> dict:
    if not state.evidence: step,name=1,"read"
    elif not state.patched: step,name=2,"patch"
    elif not state.verified: step,name=3,"test"
    else: return {"kind":"final","status":"completed"}
    return {"kind":"tool","call":{"id":f"op-{step}","tool":name,"args":{"scope":"toy-project"}}}


def run(max_turns: int=6):
    if not isinstance(max_turns,int) or isinstance(max_turns,bool) or max_turns<=0:
        raise ValueError("轮次预算必须为正整数")
    state=State(); executor=Executor({"read","patch","test"})
    status="budget_exhausted"
    for _ in range(max_turns):
        proposal=fake_model(state)
        if proposal["kind"]=="final":
            status="completed" if state.verified else "verification_missing"
            break
        call=proposal["call"]
        result=executor.execute(call,state)
        state.events.append({"call":call,"result":result})
    return state,executor,status


def self_test() -> None:
    state,ex,status=run()
    assert status=="completed" and state.verified and ex.write_count==1
    patch={"id":"op-2","tool":"patch","args":{"scope":"toy-project"}}
    assert ex.execute(patch,state)["replayed"] and ex.write_count==1
    try:
        ex.execute({"id":"x","tool":"publish","args":{"scope":"toy-project"}},state)
    except PermissionError:
        pass
    else:
        raise AssertionError("未授权工具应被拒绝")
    assert run(1)[2]=="budget_exhausted"


def run_demo() -> dict:
    state,ex,status=run()
    return {"性质":"模拟模型与内存工具,不代表真实Agent性能",
            "最终状态":status,"实际模拟写入次数":ex.write_count,"事件":state.events}


if __name__ == "__main__":
    import json
    self_test()
    print(json.dumps(run_demo(),ensure_ascii=False,indent=2))

运行结果示例

以下结果由随包脚本产生。它们不是LLM或产品性能数据。

{
  "性质": "模拟模型与内存工具,不代表真实Agent性能",
  "最终状态": "completed",
  "实际模拟写入次数": 1,
  "事件": [
    {
      "call": {
        "id": "op-1",
        "tool": "read",
        "args": {
          "scope": "toy-project"
        }
      },
      "result": {
        "status": "ok",
        "evidence": "toy:T1"
      }
    },
    {
      "call": {
        "id": "op-2",
        "tool": "patch",
        "args": {
          "scope": "toy-project"
        }
      },
      "result": {
        "status": "ok",
        "version": "toy:v2"
      }
    },
    {
      "call": {
        "id": "op-3",
        "tool": "test",
        "args": {
          "scope": "toy-project"
        }
      },
      "result": {
        "status": "passed"
      }
    }
  ]
}

输入关键词即可检索。

↑ ↓ 选择 Enter 打开Esc 关闭 · 仅本地搜索

阅读设置

阅读设置仅保存在本机浏览器,不上传任何数据。