从 0 搭一个能用的 Agent 循环

副标题：Agent 与上下文工程 · 硬核实战

本文为 vergex.cn 原创首发，转载请注明出处。

凌晨两点，你的 Agent 把一次查天气的调用循环了 47 次，最后以 “我觉得我已经回答你了” 自我终止。问题不在 prompt，而是循环本身没有定义清楚什么叫“一步”、什么叫“结束”。这篇文章把我们从零写到一个能上线的 Agent 循环的关键细节摊开讲，完整框架代码见站点下载。

1. 把循环画成状态机，而不是 prompt 拼接

大多数 Demo 把 Agent 写成一个 while 里不断拼字符串的函数，跑通三个工具就觉得万事大吉。一旦上量，你会立刻遇到两个麻烦：第一，模型某次回复不带工具调用，你却继续让它“再想想”；第二，某次工具调用失败后，模型不知道失败是“可重试”还是“该放弃”。根本原因是代码里没有显式状态，只有隐式的 messages 长度。

一个能用的最小状态机只有四个状态：

• THINK：LLM 正在决定下一步；
• TOOL_CALL：拿到工具调用，准备执行；
• OBSERVE：拿到执行结果，写回上下文；
• DONE：产生最终面向用户的回复，循环结束。

每次从 LLM 拿到响应，先解析它是不是调用了工具。如果是，进 TOOL_CALL；如果不是，直接进 DONE。不要试图在“无工具调用”时还继续追问模型“你确定吗？”——那会把模型逼进胡言乱语。终止条件必须写在代码里，而不是靠模型自觉。

代码上，我会先定义一个枚举和一个极简循环骨架：

from enum import Enum, auto

class AgentState(Enum):
    THINK = auto()
    TOOL_CALL = auto()
    OBSERVE = auto()
    DONE = auto()

class Agent:
    def __init__(self, llm_client, tools):
        self.client = llm_client
        self.tools = tools
        self.messages = []
        self.state = AgentState.THINK

    def run(self, user_input: str, max_steps: int = 10) -> str:
        self.messages.append({"role": "user", "content": user_input})
        for step in range(max_steps):
            if self.state == AgentState.THINK:
                self._think()
            elif self.state == AgentState.TOOL_CALL:
                self._execute_tools()
            elif self.state == AgentState.OBSERVE:
                self._observe()
            elif self.state == AgentState.DONE:
                break
        return self.messages[-1]["content"]

这个骨架的好处是：所有和 prompt 工程无关的边界条件（最大步数、状态跳转）都被显式捕获，而不是散落在字符串模板里。

2. 消息协议：用结构化 roles，不要用“AI 说/人类说”字符串

OpenAI SDK 1.x（以及兼容它的 vLLM、Ollama 服务）对 messages 的 role 有严格要求：system、user、assistant、tool。很多自己搭的循环会偷懒，把工具返回结果也塞进 assistant 或 user 的 content 里，短期能跑，长期一定出问题：模型会搞混“谁说了什么”，并且一旦迁移到需要 tool_call_id 的 API，就会报字段缺失。

正确的约定是：

• system：只放一次，描述 Agent 的人格、可用工具、输出格式约束；
• user：只放真实用户输入；
• assistant：只放模型自己的回复，包括 tool_calls 字段；
• tool：只放工具执行结果，每个结果必须带对应的 tool_call_id。

初始化 system message 时，建议同时给模型一个“结束信号”。例如要求它在不需要工具时直接给出最终答案，不要加任何内部思考前缀。这个信号可以显著降低无意义循环的概率。

SYSTEM_PROMPT = """你是一个帮助用户完成任务的 Agent。
你可以调用提供的工具。当你已经获得足够信息时，
直接给出最终答案，不要添加“思考：”等前缀。
"""

def _think(self):
    response = self.client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": SYSTEM_PROMPT}] + self.messages,
        tools=self.tools.schema(),
        tool_choice="auto",
        parallel_tool_calls=False,
    )
    msg = response.choices[0].message
    self.messages.append(msg.model_dump(exclude_unset=True))
    if msg.tool_calls:
        self.pending_tool_calls = msg.tool_calls
        self.state = AgentState.TOOL_CALL
    else:
        self.state = AgentState.DONE

这里有两个细节：第一，model_dump(exclude_unset=True) 会把空字段去掉，避免历史消息里堆积一堆 null；第二，parallel_tool_calls=False 在调试期非常有用，它让模型一次只调一个工具，降低你排查结果映射时的复杂度。生产环境可以视情况打开。

3. 工具注册：让 Python 函数自己生成 JSON Schema

手写 JSON Schema 容易出错，而且一旦函数签名改了，schema 就过期。用类型注解和装饰器自动生成 schema，是维护超过五个工具的底线。 核心思路是：装饰器读取函数的 __annotations__ 和 docstring，输出 OpenAI 工具格式。

import inspect
import json
from typing import Any, Callable, get_type_hints

def tool(name: str | None = None, description: str | None = None):
    def decorator(func: Callable[..., Any]):
        tool_name = name or func.__name__
        tool_desc = description or (inspect.getdoc(func) or "")
        hints = get_type_hints(func)
        hints.pop("return", None)
        properties = {}
        required = []
        for param_name, param_type in hints.items():
            properties[param_name] = {"type": "string"}  # 简化示例，实际可映射类型
            required.append(param_name)
        func._tool_schema = {
            "type": "function",
            "function": {
                "name": tool_name,
                "description": tool_desc,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required,
                },
            },
        }
        return func
    return decorator

实际生产里，类型映射表应该覆盖 str、int、float、bool、enum、list[str] 等。Pydantic 也可以做这件事，但如果你的 Agent 本身不想引入太多依赖，自己维护一个 TYPE_MAP 反而更清晰。这里我不展开完整映射，站点下载的完整框架代码里包含了一个带类型校验和默认值处理的版本。

4. 执行与回写：一步只做一次生成

Agent 最大的性能陷阱不是模型慢，而是“生成—调用—生成”被错误地合并成一次请求。比如有些实现会在拿到工具结果后，不把结果写回 messages，而是直接拼进下一次 prompt。这会导致模型看不到自己的 tool_call，也看不到上一次的 tool_call_id，最终要么报错，要么重复调用。

每一步必须严格遵守：一次 LLM 调用只产生一个 assistant 消息；每个 tool 消息都对应前一个 assistant 消息里的 tool_call_id。

def _execute_tools(self):
    observations = []
    for tc in self.pending_tool_calls:
        func = self.tools.get(tc.function.name)
        try:
            args = json.loads(tc.function.arguments)
            result = func(**args)
        except Exception as e:
            result = {"error": str(e), "retryable": True}
        observations.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": json.dumps(result, ensure_ascii=False),
        })
    self.messages.extend(observations)
    self.state = AgentState.OBSERVE

def _observe(self):
    # 回写后再次进入思考状态
    self.state = AgentState.THINK

注意我把错误结果也按同样格式写回，而不是打印到控制台。模型只有在看到 {"error": ..., "retryable": true} 时，才能决定是修正参数再调一次，还是放弃并向用户解释。如果工具抛异常你却吞掉，模型拿到空内容会开始 hallucinate 一个结果。

5. 上下文压缩：别让工具返回值撑爆窗口

工具返回常常很大。一个数据库查询可能返回几百行，一个网页抓取可能返回几万 token。如果不加处理，三轮调用之后上下文就满了。“能用的 Agent” 必须有一个显式的 token 预算管理，而不是依赖模型的隐式截断。

我的做法是给历史消息设置一个预算表：

核心规则是：assistant 的 tool_calls 和 tool 的结果必须成对保留。 如果你只删 tool 结果而保留 tool_calls，OpenAI API 会报错；如果你只删 tool_calls 而保留 tool 结果，模型会看到一堆没有来由的观察。

当总 token 超过预算时，我采用“滑动窗口 + 摘要”混合策略：先尝试删掉最旧且无关的 tool 对；如果还不够，把更早的多轮对话压缩成一段 system 追加的摘要。摘要本身也由一次独立的 LLM 调用生成，但那次调用不计入主循环。

6. 避坑：工具调用 ID 必须严格对齐

这是自己写循环时最容易踩、但官方文档不会替你强调的坑。每一个 tool 消息都必须带有正确的 tool_call_id，并且顺序要和 assistant 消息里的 tool_calls 一致。 如果你用列表推导或者异步并发执行工具，很容易出现 ID 错位。

有一次我把工具执行改成了 asyncio.gather 提速，结果返回顺序和调用顺序不一致，导致模型把天气查询的结果当成股票查询的结果，连续三轮都在解释“为什么北京的气温会影响股价”。修复方法很简单：执行前先把 (tool_call_id, function, args) 绑定到一个 dict，返回后再按原始顺序重组。

另一个常见错误是：模型某次没有调用工具，你却往 messages 里塞了一条 role="tool"。这直接触发 API 校验错误。每次回写前都要检查 msg.tool_calls 是否存在且非空。

7. 完整循环：一段可直接运行的骨架

下面是把前面片段拼起来的可运行骨架。你需要设置环境变量 OPENAI_API_KEY，并安装 openai>=1.40.0。

pip install "openai>=1.40.0"

import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def get_weather(city: str) -> str:
    """查询城市天气"""
    return f"{city} 今天晴，25°C"

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                },
                "required": ["city"],
            },
        },
    }
]

def agent_loop(user_input: str, max_steps: int = 5) -> str:
    messages = [{"role": "user", "content": user_input}]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            parallel_tool_calls=False,
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump(exclude_unset=True))

        if not msg.tool_calls:
            return msg.content

        for tc in msg.tool_calls:
            if tc.function.name == "get_weather":
                args = json.loads(tc.function.arguments)
                result = get_weather(**args)
            else:
                result = {"error": "unknown tool"}
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result, ensure_ascii=False),
            })
    return messages[-1]["content"]

if __name__ == "__main__":
    print(agent_loop("上海今天天气怎么样？"))

这个骨架可以跑通，但离生产还差三步：把 get_weather 换成真正的工具注册表、加上 token 预算管理、加上 trace 日志。站点上的完整版已经把这些补全。

8. 观测与调试：给每次循环一个 trace_id

Agent 的 bug 通常不是某一行代码错，而是多轮交互里的状态漂移。没有 trace，你根本没法复盘。我的习惯是给每次 run 分配一个 trace_id，然后把每一轮的状态、token 消耗、工具调用耗时写进结构化日志。

import uuid
import logging

logger = logging.getLogger("agent")

def run(self, user_input: str, max_steps: int = 10) -> str:
    trace_id = uuid.uuid4().hex[:12]
    self.messages.append({"role": "user", "content": user_input})
    for step in range(max_steps):
        logger.info(
            "agent.step",
            extra={
                "trace_id": trace_id,
                "step": step,
                "state": self.state.name,
                "msg_count": len(self.messages),
            },
        )
        ...

如果你用 Sentry、Langfuse 或者自研的 trace 平台，直接把 trace_id 带过去即可。建议同时记录 prompt_tokens、completion_tokens、total_tokens，以及每个工具调用的 latency。这些数据是后期做预算优化和识别“循环调用”模式的基础。

反模式清单

• 把工具返回藏进 user 消息：会破坏 role 语义，后续迁移到严格 API 时全崩。
• 不限制最大步数：模型可能无限自我调用，直到账单爆炸。
• 吞掉工具异常：模型会基于空内容编造结果， quiet failure 比报错更危险。
• 允许删除不成对的 tool/tool_call 消息：API 会拒绝，或者模型会混淆。
• 把“是否结束”交给模型自己判断：终止条件必须写在状态机里。
• 一次请求里同时生成多个步骤：每步一次 LLM 调用，便于回滚和观测。
• 上线前不做 token 预算测试：用长工具返回值压测三轮，看是否会截断关键信息。

核心总结：Agent 循环的可维护性不取决于 prompt 多精巧，而取决于状态、消息协议和工具边界是否被代码显式约束。