跳到主要内容
blog-hexo-astro
返回

Agent Loop:一个循环,就是模型与真实世界之间的全部距离(Agent架构实操一)

来源:多模态智能体 原文链接:https://mp.weixin.qq.com/s/HaWq1UZemBkWMElrFKoQuw

Table of contents#

Open Table of contents

核心结论#

Agent 的核心代码只有一个 while True 循环,整个决策树只有两个分支:模型调工具了?是就执行并喂回去继续循环,否就退出。循环本身没有智能——智能全在模型里,循环只是它的手脚。

理解 Agent 的关键是区分两件事:模型决策和 Harness 执行。模型负责”做什么”(选择工具、构造参数),Harness 负责”怎么做”(实际执行 shell 命令、读写文件)。两者通过 messages 列表连接——模型的输出成为 Harness 的输入,Harness 的执行结果成为模型下一轮的输入。

整个 Agent = 模型决策 + Harness 执行 + 消息历史传递,三件事加在一起就是 Agent。不需要 Planner、Memory、Orchestrator 等复杂概念,那些都是在这三件事基础上叠加的能力。从这篇开始,我们逐层叠加:下一篇 给循环加上工具分发表,后续 叠加 权限和Hooks,一路到 多 Agent 协作。

Agent Loop 流程图 Agent Loop 流程:一个 while True 循环,模型调工具→执行→喂回→再问,不调工具就停

关键洞察#

洞察一:Agent Loop 做的事,就是把你手动复制粘贴的工作自动化#

没有 Agent 之前,你让模型分析文件的工作流是这样的:

  1. 你输入:“帮我看看这个项目有哪些 Python 文件”
  2. 模型输出:“请运行 find . -name '*.py'”
  3. 你切到终端,执行命令
  4. 你把输出复制粘贴回对话框
  5. 模型看到结果,继续推理

每一次来回你都在做同一件事:把工具结果手动传回给模型。Agent Loop 的唯一贡献就是把这个中间层自动化——循环自动执行模型指定的工具,自动把结果追加回 messages,自动再问一次模型。人类的角色从”搬运工”变成了”监督者”。

洞察二:退出条件不是”模型说完了”,而是”模型没有调工具”#

这两件事不等价。模型可以:

我们只关心它有没有举手要工具。这是理解整个循环的关键:tool_calls 是否非空 就继续,否则退出。模型输出的文本内容不决定循环是否继续,只有工具调用才决定。

洞察三:智能在模型里,循环只是手脚#

循环是机械的,没有任何智能。它只负责四件事:

  1. 把 messages + tools 定义发给模型
  2. 等模型返回结果
  3. 如果有工具调用,执行工具并把结果追加回 messages
  4. 回到第 1 步

真正的决策——调不调工具、调哪个工具、传什么参数——全在模型里。循环只是”手脚”,负责把模型的决策执行出来。这也解释了为什么 Agent 的核心循环代码量可以这么少:它真的只做机械的搬运工作。

洞察四:30 行 vs 1729 行的差距不是”功能多”,而是”保护多”#

Claude Code 的核心文件 query.ts 有 1729 行,但核心循环逻辑和我们这 30 行本质上一样。剩下的 1700 行全是保护机制:

保护机制作用
状态恢复(checkpoint/restore)崩溃后从上次检查点恢复,不从头来
Context 压缩messages 太长时自动摘要,防止超出上下文窗口
Token 截断重试模型输出被截断时自动续写,不丢内容
流式并行执行模型还在生成时就开始执行工具,减少延迟
多路退出策略不只是”不调工具就退出”,还有 maxTurns、用户中断、错误恢复等

生产级 Agent 和实验级 Agent 的差距,不在于核心循环多复杂,而在于”出问题时能不能继续跑”。

五步执行流程详解#

步骤动作代码实现注意事项
1. 用户输入追加到 messages 列表messages.append({"role": "user", "content": query})用户输入永远是 role=“user”
2. 调用模型messages + TOOLS 一起发给 APIclient.chat.completions.create(model=MODEL, messages=..., tools=TOOLS)TOOLS 是预先定义的 JSON Schema
3. 追加回答把 Pydantic 对象转为纯字典后 appendmsg = {"role": am.role, "content": am.content}必须转 dict,Pydantic 模型直接塞会报错
4. 判断退出检查 tool_calls 是否非空if not am.tool_calls: return不关心文本内容,只看有没有工具调用
5. 执行工具逐个执行并收集结果handler(**json.loads(tc.function.arguments))每个结果通过 tool_call_id 对应到具体调用

完整核心函数不到 30 行,但覆盖了 Agent 的全部基础能力。

一个完整的场景推演#

假设用户输入 “帮我创建一个 hello.py 文件并运行它”,Agent Loop 的执行过程:

Round 1:
  用户: "帮我创建一个 hello.py 文件并运行它"
  模型: tool_calls=[write_file(path="hello.py", content="print('Hello!')")]
  执行: write_file → "Wrote 24 bytes to hello.py"
  追加: messages += [assistant_msg, tool_result]

Round 2:
  模型(看到历史): tool_calls=[bash(command="python hello.py")]
  执行: bash → "Hello!"
  追加: messages += [assistant_msg, tool_result]

Round 3:
  模型(看到完整历史): content="文件已创建并运行成功,输出 Hello!"
  没有 tool_calls → 退出循环

注意每一轮模型都能看到完整的对话历史——包括之前的用户输入、模型回答、工具调用和工具结果。这就是为什么 messages 列表如此重要:它是 Agent 的”工作记忆”。

Anthropic 格式 vs OpenAI 格式#

虽然底层信号一样(“模型有没有举手要工具”),但两个 SDK 的数据格式不同:

概念Anthropic 格式OpenAI 格式
判断是否调工具response.stop_reason == "tool_use"tool_calls 是否非空
工具调用 ID 字段tool_use_idtool_call_id
工具结果的 role嵌在 user 消息里(content 数组中的 tool_result 块)独立的 tool 消息
工具调用块格式content 里的 type: "tool_use" 块message.tool_calls 数组
消息追加方式assistant 消息直接 append必须转为纯字典再 append

形式不同,但本质信号一样。理解了这一点,切换 SDK 只是改数据格式,核心逻辑不变。

Claude Code 的关键差异#

循环继续的判断方式不同。我们的代码靠 tool_calls 是否非空,CC 用 needsFollowUp 标志位,在流式接收时只要检测到 tool_use 块就设为 true。因为流式响应里 stop_reason 可能滞后——内容块已到,元数据还没更新,只看 stop_reason 在流式场景下可能漏判。

State 对象有 10 个字段 vs 我们只有 messages。比较有意思的两个:maxOutputTokensRecoveryCount(token 截断重试,上限 3 次——模型输出被截断不能当”做完了”)和 pendingToolUseSummary(后台用 Haiku 生成工具调用摘要,压缩 context 长度省 token)。

退出路径从 1 条变成 7+ 条。我们的代码只有一条路(模型不调工具就退出),CC 覆盖了:达到 maxTurns、context 超长(尝试压缩后继续)、token 截断(恢复后继续)、模型报错、用户中断、Stop Hook 拦截、token budget 耗尽……每条路径都有独立的处理逻辑和恢复策略。

流式并行工具执行。CC 的 StreamingToolExecutor 在模型还在生成响应时,对已到达的工具调用就开始执行——不等模型说完。根据工具的 concurrency-safe 标记决定并发还是串行。纯性能优化,但在大量工具调用的任务里能显著减少总延迟。

完整可运行代码#

"""s01_agent_loop.py — Agent 核心循环:30 行代码实现一个能用工具的 AI Agent

运行前准备:
  1. pip install openai
  2. export DEEPSEEK_API_KEY="你的密钥"
  3. python s01_agent_loop.py

这个脚本演示了 Agent 的最简形态:一个 while True 循环,
模型决定调什么工具,循环负责执行并把结果喂回去。
"""
import json, os, subprocess
from openai import OpenAI

# ===== 模型配置 =====
# 使用 DeepSeek API(兼容 OpenAI SDK)
# 也可以换成 OpenAI:client = OpenAI()  MODEL = "gpt-4o"
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),  # 从环境变量读取 API 密钥
    base_url="https://api.deepseek.com"      # DeepSeek 的 API 地址
)
MODEL = "deepseek-v4-flash"  

# ===== System Prompt =====
# 告诉模型它是谁、能做什么
SYSTEM = "You are a helpful assistant with access to a bash shell. Use tools to help the user."

# ===== 工具定义(JSON Schema 格式)=====
# 模型通过这些 Schema 知道自己有哪些工具可用、每个工具接受什么参数
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "bash",
            "description": "Run a shell command.",  # 模型靠这个描述决定什么时候用这个工具
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "Shell command to execute"
                    }
                },
                "required": ["command"],  # command 是必填参数
            },
        },
    }
]

# ===== 安全黑名单 =====
# 这是最粗粒度的安全措施:某些危险命令直接拒绝执行
# 生产环境会用更复杂的权限系统(见 s03)
DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]


def run_bash(command: str) -> str:
    """执行 shell 命令并返回输出

    三层安全检查:
    1. 黑名单匹配:包含危险关键词直接拒绝
    2. 超时保护:30 秒超时防止死循环命令
    3. 异常捕获:任何错误都返回错误信息而不是崩溃
    """
    # 安全检查:遍历黑名单
    for d in DENY_LIST:
        if d in command:
            return f"Error: Dangerous command blocked ('{d}')"
    try:
        # subprocess.run 执行命令,capture_output 捕获 stdout 和 stderr
        result = subprocess.run(
            command,
            shell=True,           # 用 shell 解析命令(支持管道、重定向等)
            capture_output=True,  # 捕获输出而不是打印到终端
            text=True,            # 返回字符串而不是 bytes
            timeout=30,           # 超时保护:防止命令卡死
        )
        # 合并 stdout 和 stderr 一起返回给模型
        return result.stdout + result.stderr
    except subprocess.TimeoutExpired:
        return "Error: Command timed out (30s limit)"


def agent_loop(messages: list):
    """Agent 核心循环 — 整个 Agent 的心脏

    循环逻辑:
    1. 把消息历史 + 工具定义发给模型
    2. 模型返回回答(可能包含工具调用)
    3. 如果有工具调用 → 执行工具 → 结果追加到消息历史 → 继续循环
    4. 如果没有工具调用 → 退出循环

    关键设计:
    - messages 是可变列表,循环内直接修改(append)
    - 每次调用模型都带上完整的 SYSTEM + messages + TOOLS
    - 工具结果通过 tool_call_id 关联到具体的工具调用
    """
    while True:
        # 调用模型:system prompt + 历史消息 + 工具定义
        response = client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "system", "content": SYSTEM}] + list(messages),
            tools=TOOLS,      # 告诉模型有哪些工具可用
            max_tokens=8000,  # 限制模型单次输出的最大 token 数
        )

        # --- 追加模型回答 ---
        assistant_msg = response.choices[0].message
        # 必须转为纯字典!Pydantic 对象直接 append 会导致后续 API 调用报错
        msg = {"role": assistant_msg.role, "content": assistant_msg.content}
        if assistant_msg.tool_calls:
            # 如果模型调了工具,把工具调用信息也转为字典格式
            msg["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": tc.type,
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments,  # JSON 字符串
                    },
                }
                for tc in assistant_msg.tool_calls
            ]
        messages.append(msg)

        # --- 退出判断 ---
        # 核心逻辑:没有工具调用 → 模型认为任务完成 → 退出
        if not response.choices[0].message.tool_calls:
            return

        # --- 执行工具 ---
        for tool_call in response.choices[0].message.tool_calls:
            # 解析模型传来的参数(JSON 字符串 → Python dict)
            command = json.loads(tool_call.function.arguments)["command"]
            # 执行工具,拿到输出
            output = run_bash(command)
            # 把工具结果追加到消息历史
            # role="tool" + tool_call_id 让模型知道这是哪个工具调用的结果
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,  # 关联到具体的工具调用
                "content": str(output),
            })
        # 循环继续:模型看到工具结果后会决定下一步做什么


if __name__ == "__main__":
    messages = []  # 消息历史:整个会话的"工作记忆"
    print("Agent Loop (输入 quit 退出)")
    while True:
        query = input("\n> ")
        if query.strip().lower() in ("quit", "exit"):
            break
        # 用户输入追加到消息历史
        messages.append({"role": "user", "content": query})
        # 进入 Agent 核心循环
        agent_loop(messages)
        # 打印模型最后一条回复(文本内容)
        last = messages[-1]
        if last.get("content"):
            print(f"\n{last['content']}")

代码运行示例#

运行后你可以试试这些交互:

> 列出当前目录的文件
# 模型会调用 bash("ls -la"),然后告诉你文件列表

> 帮我创建一个 README.md
# 模型会调用 bash("echo '# My Project' > README.md"),然后确认创建成功

> 这个目录下有几个 Python 文件?
# 模型会调用 bash("find . -name '*.py' | wc -l"),然后告诉你数量

每次交互,Agent Loop 都会自动执行工具并把结果喂回给模型,你不需要手动复制粘贴。

当前实现的局限#

局限问题解决方案
只有一个工具 bash读文件靠 cat,写文件靠 echo >,路径有空格就出问题下一篇 加真正的 read_file、write_file、edit_file、glob
没有权限控制任何命令都能执行,rm -rf / 只靠字符串匹配后续 加三道闸门权限系统
工具分发是硬编码加新工具要改 agent_loop 函数下一篇 引入 TOOL_HANDLERS 分发表

循环本身不需要改动——这正是分发表设计的价值。下一篇 工具分发表 会给它真正的文件操作工具,同时让”加一个工具”变成”加一行代码”的事。

学习笔记#

记录自己的理解和延伸思考

Agent 的本质是什么? 读完这篇最大的感触是:Agent 并没有什么神秘的。它就是一个 while True 循环,不断问模型”你要做什么”,然后帮它执行。真正的智能在模型里,循环只是执行层。这让我想到一个类比:Agent Loop 就像一个人的手脚,模型就像大脑。手脚只是执行大脑的指令,但没有手脚,大脑再聪明也无法与世界交互。

Harness 的重要性被低估了。大多数 AI 教程关注”怎么让模型更聪明”(prompt engineering、fine-tuning),但很少有人关注”怎么让模型的输出真正执行起来”。Claude Code 的 1729 行代码里,核心循环只有 30 行,剩下 1700 行都是 Harness 层的保护机制。这说明生产级 Agent 的难点不在”让模型做对”,而在”模型做错时怎么办”。

这个系列的演进路线。从 s01 到 s15,本质上就是在核心循环上不断叠加保护机制:s02 加工具分发,s03 加权限,s04 加 Hook 解耦,s05 加任务规划,s06 加子 Agent,s07 加技能加载,s08 加上下文压缩,s09 加持久化记忆……每一篇都在回答同一个问题:“模型做错了/做不了/做太多时怎么办”。



上一篇
如何在本博客写一篇新笔记:完整教程
下一篇
如何搭建本地开发流水线