单 Agent 最小实现：用状态机控制决策循环

上一篇认识了 Agent 的模型、工具、记忆与循环。本篇把范围进一步收窄：只实现一个单 Agent，并重点解决“什么时候继续、什么时候停止”。示例使用离线的规则模型模拟模型决策，因此不需要密钥或网络，能够先验证 Agent 的控制结构，再替换成真实模型。

单 Agent 到底在循环什么

单 Agent 不是把所有逻辑都交给模型，而是由一个模型反复处理同一个任务状态。每一轮可以抽象为四步：

把目标、历史和工具结果交给模型，请它提出下一步动作。
检查动作是否符合应用定义的格式和允许范围。
执行工具，把成功结果或错误写回状态。
如果模型已经给出最终答案，或者触发停止条件，就结束；否则进入下一轮。

这里的“单”指只有一个负责决策的模型循环，不代表只能有一个工具。与普通函数调用相比，Agent 的关键新增部分是状态迁移：上一轮的观察结果会影响下一轮决策。循环控制权必须留在 Python 程序中，不能只在提示词里要求模型“适时停止”。

先定义动作和工具

为了让动作可检查，先约定模型只能返回两种字典：tool_call 表示调用工具，final 表示完成任务。工具也使用白名单注册，模型返回的名字不能直接当作 Python 函数执行。

from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class State:
    goal: str
    history: list[dict[str, Any]] = field(default_factory=list)
    steps: int = 0


def lookup_weather(city: str) -> str:
    data = {"北京": "晴，26°C", "上海": "多云，28°C"}
    if city not in data:
        return f"没有找到 {city} 的示例天气"
    return data[city]


TOOLS: dict[str, Callable[..., str]] = {
    "lookup_weather": lookup_weather,
}

State 是短期记忆：goal 不变，history 记录动作和观察，steps 记录已经消耗的循环次数。生产代码还可以加入请求 ID、用户权限和超时信息，但这些字段都应由应用维护，而不是由模型自行决定。

用离线模型模拟一次决策

下面的函数代替真实模型，只处理“查询某城市天气”这个固定目标。真实接入 SDK 后，仍建议把 SDK 返回值转换成同样的动作协议，这样循环本身不必和某一家模型服务绑定。

def decide(state: State) -> dict[str, Any]:
    """离线模拟模型：最多提出一次天气查询，然后给出答案。"""
    if not state.history:
        return {
            "type": "tool_call",
            "name": "lookup_weather",
            "arguments": {"city": "北京"},
        }

    observation = state.history[-1]
    if observation["type"] == "tool_result":
        return {"type": "final", "content": f"查询结果：{observation['result']}"}
    return {"type": "final", "content": "无法继续处理。"}

注意，decide 返回的是数据，不是可执行代码。模型即使返回了未知工具、缺少参数或错误类型，应用也应该拒绝执行。把“模型建议”和“程序动作”分开，是后续加入参数校验、日志和人工审批的基础。

实现带边界的 Agent 循环

接下来实现执行器。它包含四类停止条件：模型主动结束、动作格式错误、工具执行失败，以及达到最大步数。最后一类是硬上限，即使模型一直请求工具，循环也不会无限运行。

def run_agent(goal: str, max_steps: int = 3) -> str:
    if max_steps < 1:
        raise ValueError("max_steps 必须至少为 1")

    state = State(goal=goal)

    while state.steps < max_steps:
        state.steps += 1
        action = decide(state)

        if not isinstance(action, dict) or action.get("type") not in {"tool_call", "final"}:
            return "停止：模型返回了无法识别的动作。"

        if action["type"] == "final":
            content = action.get("content")
            return content if isinstance(content, str) else "停止：最终答案格式错误。"

        name = action.get("name")
        arguments = action.get("arguments", {})
        tool = TOOLS.get(name)
        if tool is None:
            return f"停止：工具 {name!r} 不在白名单中。"
        if not isinstance(arguments, dict):
            return "停止：工具参数必须是对象。"

        try:
            result = tool(**arguments)
        except (TypeError, ValueError) as exc:
            state.history.append({"type": "tool_error", "error": str(exc)})
            return f"停止：工具参数或执行失败：{exc}"

        state.history.append({
            "type": "tool_result",
            "name": name,
            "result": result,
        })

    return f"停止：达到最大步数 {max_steps}，未得到最终答案。"


if __name__ == "__main__":
    print(run_agent("查询北京天气"))

执行器每次循环只允许发生一个明确动作：返回最终文本，或者从白名单中取出一个工具并执行。工具结果写回 history 后，下一轮才能看到它。若把 max_steps 设为 1，示例会在工具调用后停止；设为 3，才有机会进入下一轮生成最终答案。这说明最大步数不是装饰参数，而是 Agent 的资源和风险边界。

为什么还要记录工具错误

工具失败时可以选择立即结束，也可以把错误作为观察结果交给模型，让模型修正参数后重试。无论选择哪种策略，都要限制重试次数，并区分可恢复错误和不可恢复错误。例如城市不存在可能允许模型改问法；权限不足、参数类型错误或外部服务持续超时，则不应无限重试。

真实项目还应记录每一轮的输入摘要、动作名称、耗时和结束原因，但不要把密钥、完整隐私数据或未经脱敏的工具返回值写入普通日志。观测数据既要帮助排错，也要遵守数据边界。

常见问题

把 max_steps 交给模型是否可以？ 不可以。模型可以建议结束，但最终上限必须由应用代码强制执行。

为什么不让模型直接调用函数？ 因为模型输出是不可信输入。白名单、参数校验和权限检查必须在函数执行前完成。

是否一定要保存数据库记忆？ 不一定。本例的列表只服务于一次运行，称为短期记忆。需要跨请求恢复任务时，再设计持久化状态，并考虑过期、并发和隐私。

单 Agent 能解决复杂任务吗？ 它适合先验证“决策—工具—观察”的闭环。任务变复杂后，可以增加工具和状态字段，但应先保持单循环可测试，再考虑多 Agent 或更复杂的工作流。

小结

单 Agent 的最小实现可以很小：一个状态对象、一个决策函数、一个工具白名单和一个受限循环。模型负责提出下一步，程序负责验证、执行和停止。先用离线决策器验证状态迁移，再接入真实 API，能够把模型质量问题与控制逻辑问题分开排查。下一步可以在此基础上学习状态、短期记忆与持久化，而不是贸然把循环扩展成没有边界的自动化程序。

搜索时光笔记