Agent Loop：多轮循环的核心执行机制 · 生产级 Agent

很多人第一次接触 Function Calling，写出来的代码是一条直线：发请求 → 模型要调工具 → 执行工具 → 再发一次请求 → 打印回复。跑"查个天气"这种一步任务没问题，一碰到需要连续几步才能完成的真实任务，程序就会在半路断掉。这篇笔记专门解决这个问题：搞清楚 Agent 最核心的执行机制——多轮循环（Agent Loop）。读完你会明白为什么必须用循环、循环靠什么停下来、流式模式下工具调用参数为什么要手动拼接，最后我们动手写一个能读写文件的最小可用 Agent。

"一次工具调用"的假设为什么不成立

先看一段典型的直线式代码，它隐含了一个致命假设：

# 直线流程：默认模型只会要求调一次工具
first = call_llm(messages)

if first.tool_calls:
    tool_output = run_tool(first.tool_calls[0])
    second = call_llm(messages + [assistant_msg, tool_msg])
    print(second.content)   # 假设这里一定是最终答案

这个假设是：模型调完一次工具，第二次响应必然是文字答案。但模型并不这么工作。

举个例子，你给 Agent 的任务是：

读一下 meeting.md，把里面所有以 TODO 开头的行抽出来，汇总写进 todo-list.md。

模型的合理执行路径至少是两步：先调读文件工具拿到 meeting.md 的内容，再调写文件工具生成 todo-list.md，最后才用自然语言汇报结果。也就是说，第二次响应里装的很可能不是答案，而是又一个 tool_calls。此时直线代码去打印 second.content，拿到的是 None 或者一句"接下来我要写入文件"的过程性描述，然后程序就退出了——任务只完成了一半。

问题的本质：需要几步才能完成任务，是模型在运行时动态决定的，写代码时根本不可能预知。 所以执行结构不能是固定次数的直线，必须是一个由模型输出驱动的循环。

循环骨架：以"是否还有 tool_calls"为唯一判据

把直线改成循环，规则只有一条：这一轮响应里带 tool_calls，就执行工具、把结果塞回消息历史、进入下一轮；直到某一轮模型不再要求调工具、给出纯文字回复，循环退出。骨架如下：

messages = [system_msg, user_msg]

while True:
    reply = call_llm(messages)          # 一轮 = 一次模型调用
    messages.append(to_assistant_msg(reply))

    if not reply.tool_calls:            # 纯文字回复 => 任务完成
        break

    for call in reply.tool_calls:       # 可能一轮里有多个调用
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": run_tool(call),  # 结果必须回填给模型
        })

有两个细节值得注意：

assistant 消息（含 tool_calls）和 tool 消息都要追加进 messages。 模型本身不保存任何状态，它之所以"记得"上一步读到了什么，完全是因为每一轮请求都把完整历史重新发了过去。漏掉任何一条，下一轮模型就是在信息缺失的状态下推理。
每条 tool 消息必须携带对应的 tool_call_id。 一轮里可能有多个工具调用，模型靠这个 id 把结果和请求对上号。

停止是工程问题：上限与三种退出方式

理想情况下，循环由模型自主终止：它认为任务完成了，就输出文字答案。但生产环境不能只依赖这个理想情况。做 Agent 有一条基本纪律——模型的一切行为都要套在工程约束之内，工程约束优先于模型自主性。

因为模型确实会失控：参数错了重试、重试又错、换个写法再试，在同一个工具上原地打转；或者两个描述含混的工具让它反复横跳。没有上限的话，最坏结果是无限循环——任务没完成，API 额度先烧光了。

所以主循环必须有一个最大轮数：

MAX_STEPS = 24

for step in range(MAX_STEPS):
    ...
else:
    print("已达最大步数，任务强制终止")

上限设多大，是在两头之间取平衡：设成个位数，稍微复杂点的任务会被无辜掐断（十几步的正常任务并不罕见）；设成几百，真失控的时候你要眼睁睁看它空转很久才停。几十这个量级是常见的取值区间，具体数字按你的任务类型调。

归纳起来，Agent 循环有三种停止方式：

停止方式	触发条件	性质
自然结束	模型输出纯文字回复，不再有 `tool_calls`	理想路径
达到上限	轮数触顶，系统强制退出	工程兜底
工具失败	失败信息以字符串形式交回模型	不直接停，由模型决定重试、换路或放弃

第三种最容易做错。工具执行失败时，正确做法不是抛异常终止程序，而是把错误信息当成普通的工具结果回传，让模型看到 Error: ... 之后自己判断下一步。失败是信息，不是终点。

流式模式下，tool_calls 是碎片，要按 index 拼

非流式调用里，tool_calls 是一次性给全的，直接用就行。但实际产品几乎都要流式输出（用户要看到文字逐字出现），而流式响应里，每个 chunk 的 delta.tool_calls 只是增量碎片。一个 read_text 调用可能被拆成这样：

chunk A:  [{index: 0, id: "tc_91x", function: {name: "read_text"}}]
chunk B:  [{index: 0, function: {arguments: "{\"rel_"}}]
chunk C:  [{index: 0, function: {arguments: "path\":\"meet"}}]
chunk D:  [{index: 0, function: {arguments: "ing.md\"}"}}]

只有把 B、C、D 里的 arguments 片段按顺序连起来，才能得到合法的 JSON 参数 {"rel_path":"meeting.md"}。拼接的锚点是 index 字段：同一个 index 的碎片属于同一个工具调用。模型一轮可能并行发出多个调用（index 为 0、1、2……），所以标准做法是用一个以 index 为键的字典分别累积，流结束后再按 index 排序还原成完整列表。下面的实现里会看到这段代码。

动手实现：会读写文件的最小 Agent

目标：给模型三个工具——列目录、读文件、写文件，让它自主完成"从 meeting.md 抽 TODO 写入 todo-list.md"这类多步任务。全程不写任何"先读后写"的业务逻辑，步骤规划完全交给模型。

接口用通用的 OpenAI 兼容格式，base_url 可配置，DeepSeek、Kimi、GLM 或任何兼容网关都能直接跑。先准备环境：

mkdir -p mini-agent/sandbox && cd mini-agent
printf "TODO 确认发布窗口\n结论：灰度比例定为 5%%\nTODO 通知运营同步文案\n" > sandbox/meeting.md
export LLM_API_KEY=你的key
export LLM_BASE_URL=https://api.deepseek.com/v1   # 换成你用的服务商
export LLM_MODEL=deepseek-chat

初始化与安全护栏

# mini_agent.py
import json
import os
from pathlib import Path

from openai import OpenAI

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ["LLM_BASE_URL"],   # 任何 OpenAI 兼容服务
)
MODEL = os.environ["LLM_MODEL"]

SANDBOX = (Path(__file__).parent / "sandbox").resolve()
SANDBOX.mkdir(exist_ok=True)

MAX_STEPS = 24


def resolve_inside_sandbox(rel: str) -> Path | None:
    """把相对路径解析到沙盒内；越界返回 None。"""
    candidate = (SANDBOX / rel).resolve()
    try:
        candidate.relative_to(SANDBOX)
    except ValueError:
        return None
    return candidate

SANDBOX 是整个 Agent 的物理边界：所有文件操作先经过 resolve_inside_sandbox()，路径先 resolve() 展开再验证是否仍在沙盒之内。模型完全可能生成 ../../.ssh/id_rsa 这样的参数——不管是幻觉还是被注入的恶意指令，护栏都必须在工具层拦住，而不是指望模型自觉。

三个工具与注册表

def list_dir() -> str:
    entries = sorted(
        str(p.relative_to(SANDBOX))
        for p in SANDBOX.rglob("*") if p.is_file()
    )
    return "\n".join(entries) or "(空目录)"


def read_text(rel_path: str) -> str:
    target = resolve_inside_sandbox(rel_path)
    if target is None:
        return "Error: 路径越出沙盒，已拒绝"
    if not target.is_file():
        return f"Error: 找不到文件 {rel_path}"
    return target.read_text(encoding="utf-8")


def write_text(rel_path: str, text: str) -> str:
    target = resolve_inside_sandbox(rel_path)
    if target is None:
        return "Error: 路径越出沙盒，已拒绝"
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(text, encoding="utf-8")
    return f"ok: 已写入 {rel_path}，共 {len(text)} 字符"


TOOL_IMPL = {
    "list_dir": list_dir,
    "read_text": read_text,
    "write_text": write_text,
}

TOOL_SCHEMAS = [
    {"type": "function", "function": {
        "name": "list_dir",
        "description": "列出沙盒目录下全部文件的相对路径，每行一个",
        "parameters": {"type": "object", "properties": {}},
    }},
    {"type": "function", "function": {
        "name": "read_text",
        "description": "读取沙盒内一个文本文件的全部内容",
        "parameters": {
            "type": "object",
            "properties": {"rel_path": {"type": "string", "description": "相对沙盒根目录的路径"}},
            "required": ["rel_path"],
        },
    }},
    {"type": "function", "function": {
        "name": "write_text",
        "description": "在沙盒内创建或覆盖一个文本文件",
        "parameters": {
            "type": "object",
            "properties": {
                "rel_path": {"type": "string", "description": "相对沙盒根目录的路径"},
                "text": {"type": "string", "description": "要写入的完整文本"},
            },
            "required": ["rel_path", "text"],
        },
    }},
]

注意所有工具的返回值都是字符串，失败也返回 Error: 开头的字符串而不是抛异常——原因前面说过：错误要作为信息流回模型。TOOL_SCHEMAS 里的 name 是模型和实现之间唯一的纽带：模型按名字发起调用，我们按名字去 TOOL_IMPL 查函数。

流式请求与碎片拼接

def stream_one_round(messages) -> tuple[str, list[dict]]:
    """发起一轮流式请求，返回 (完整文字, 拼接好的工具调用列表)。"""
    resp = client.chat.completions.create(
        model=MODEL, messages=messages,
        tools=TOOL_SCHEMAS, stream=True,
    )
    text_parts: list[str] = []
    pending: dict[int, dict] = {}       # index -> 累积槽位

    for chunk in resp:
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta

        if delta.content:               # 文字边收边打印
            print(delta.content, end="", flush=True)
            text_parts.append(delta.content)

        for piece in delta.tool_calls or []:
            slot = pending.setdefault(
                piece.index, {"id": "", "name": "", "args": ""}
            )
            if piece.id:
                slot["id"] = piece.id
            if piece.function and piece.function.name:
                slot["name"] += piece.function.name
            if piece.function and piece.function.arguments:
                slot["args"] += piece.function.arguments

    calls = [pending[i] for i in sorted(pending)]
    return "".join(text_parts), calls

主循环与工具分发

def run_tool(name: str, raw_args: str) -> str:
    fn = TOOL_IMPL.get(name)
    if fn is None:
        return f"Error: 不存在的工具 {name}"
    try:
        kwargs = json.loads(raw_args) if raw_args else {}
    except json.JSONDecodeError:
        return f"Error: 参数不是合法 JSON: {raw_args}"
    try:
        return fn(**kwargs)
    except Exception as exc:            # 工具崩溃不能拖垮循环
        return f"Error: 工具异常 {type(exc).__name__}: {exc}"


def solve(task: str) -> None:
    messages = [
        {"role": "system",
         "content": "你是文件助理，只能通过给定工具操作沙盒内的文件。"},
        {"role": "user", "content": task},
    ]

    for step in range(1, MAX_STEPS + 1):
        print(f"\n[第 {step} 轮]")
        text, calls = stream_one_round(messages)

        assistant_msg: dict = {"role": "assistant",
                               "content": text or None}
        if calls:
            assistant_msg["tool_calls"] = [
                {"id": c["id"], "type": "function",
                 "function": {"name": c["name"], "arguments": c["args"]}}
                for c in calls
            ]
        messages.append(assistant_msg)

        if not calls:                   # 停止方式一：自然结束
            print()
            return

        for c in calls:
            output = run_tool(c["name"], c["args"])
            print(f"\n  -> {c['name']}({c['args']})")
            print(f"  <- {output[:120]}")
            messages.append({           # 停止方式三藏在这里：
                "role": "tool",         # 失败信息同样回填，交模型定夺
                "tool_call_id": c["id"],
                "content": output,
            })

    print(f"\n[已达上限 {MAX_STEPS} 轮，强制停止]")   # 停止方式二


if __name__ == "__main__":
    print("最小 Agent（输入 q 退出）")
    while True:                         # 外层：对话循环，一人一问
        task = input("\n> ").strip()
        if task.lower() in {"q", "quit"}:
            break
        if task:
            solve(task)                 # 内层：Agent 循环，一问多轮

留意最后这两层循环的分工：外层 while 是对话循环，节奏由用户输入驱动，一次迭代对应一个任务；内层 for 是 Agent 循环，节奏由模型输出驱动，一次迭代对应一次模型调用加一批工具执行。两者概念完全不同，上限约束只加在内层。

运行 python mini_agent.py，输入前面那个 TODO 汇总任务，可以看到模型自己规划出 read_text → write_text → 文字汇报的完整链路。想观察循环的推进过程，可以在每轮开头打印 len(messages)：每多一轮，历史里就多出一条 assistant 消息和至少一条 tool 消息，直到最后一轮模型只产出文字，循环收口。

实践要点与常见坑

content 为 None 不是 bug。 模型专注发起工具调用的那几轮，往往不带文字。判断任务是否结束的依据永远是"有没有 tool_calls"，不是"有没有 content"。
参数 JSON 偶尔会坏。 流式拼接出来的 arguments 理论上是合法 JSON，但模型偶发生成不规范内容。解析必须包 try/except，失败时把错误信息回传给模型让它重发，而不是让进程崩溃。
频繁触顶先查工具描述。 如果 Agent 经常跑满上限还没结果，多数时候不是上限太小，而是工具的 description 写得含混，模型选不对工具或传不对参数。
安全校验只能放在工具层。 路径越权、危险操作的拦截要写在工具实现里，system prompt 里的"请不要访问沙盒外的文件"只是建议，不是防线。
消息历史一条都不能少。 tool_call_id 对不上或漏回填 tool 消息，多数 OpenAI 兼容服务会直接报 400，这是新手最高频的报错来源。

本系列后面还会在这个循环骨架上继续叠加：把工具集抽象成可插拔的注册器、接入 MCP 外部工具、用 skills 组织更复杂的能力。但不管上层怎么长，核心执行机制都是这几十行循环。

动手清单

跑通本文的最小 Agent，用 TODO 汇总任务验证"读 → 写 → 汇报"的多轮链路，并打印每轮的 messages 长度，观察历史如何增长。
故意让任务失败一次：要求模型读一个不存在的文件，观察它拿到 Error: 字符串后如何自我修正（通常会先调 list_dir 再重试）。
把 MAX_STEPS 改成 2，跑一个需要三步以上的任务，体验"达到上限强制停止"这条兜底路径。
给沙盒外的路径下手：输入"读一下 ../mini_agent.py"，确认路径护栏能拦住并把拒绝信息回传给模型。
加一个 append_text 追加写入工具（含 schema、实现、注册三处），验证注册表模式下扩展一个工具需要改动的最小面积。

*本文为学习笔记，知识框架参考自叶小钗「生产级 Agent」系列课程；文字与代码示例为本站原创整理。*