从一次模型调用到 Agent：我今天理解了控制循环

今天开始系统学习 Python 和 Agent 开发。

我有多年 Rails 后端开发经验，对 API、数据库、业务流程和后台任务都比较熟悉。刚接触 Agent 时，我把它理解成“能力更强、可以连续思考的大模型”。

看完一段最小实现的代码后，我发现这个理解不对。

我原来的理解错在哪

以前看到 Agent “思考、调用工具、继续思考”，我以为连续性来自模型自身，好像它一直在那里运转。

代码里却不是这样：每次模型调用都是一次独立的请求，由 Runner 控制下一次调用。模型之所以“记得”上一轮查到的订单数据，只是因为 Python 把工具结果追加到 messages，然后带着完整的历史再调一次模型。

我把 mock_model 换成打印 messages 内容的代码，跑了一遍就看清楚了：模型每次只对当前收到的上下文做一次响应。它没有连续性，连续性来自外面的循环。

Agent 的连续行动，来自程序不断调用模型形成的控制循环。

模型负责判断下一步需要做什么，Python 程序负责校验、执行、记录状态以及决定何时结束。

想明白“数据如何进入上下文、程序如何控制下一步”这两个问题后，RAG、Memory、Tool Calling、多 Agent 这些概念突然都变得好理解了：

RAG 会把检索结果加入上下文。
Memory 会在调用模型前取回相关记录。
Tool Calling 会把函数执行结果加入上下文。
多 Agent 会把另一个 Agent 的输出作为新的输入。
Runner 负责组织这些信息并推进执行。

从订单查询开始

假设用户问：

帮我看看 ORDER-1001 现在是什么状态，然后告诉我怎么回复客户。

模型本身看不到业务数据库，也无法凭空知道订单状态。我让 AI 帮我写了个最小实现，运行过程分为这几步：

Python 将用户问题发送给模型。
模型判断需要调用 get_order。
Python 检查工具名称和参数。
Python 执行订单查询。
查询结果被加入消息列表。
Python 再次调用模型。
模型根据订单数据生成最终回复。
Python 结束循环并返回结果。

我把这 8 步在纸上画了一遍才发现：模型只出现在第 2、6、7 步，其他每一步都是 Python 在干活。模型输出的工具名称和参数只是一份执行建议，执行权在 Python 程序手里。

分工其实很干脆：

模型：提出下一步动作
程序：控制动作是否能够执行
工具：访问真实业务数据或外部系统

用 Rails 思维理解 Agent

在 Rails 项目里，我熟悉的业务流程是提前写死的：

order = Order.find_by!(number: order_number)
policy = RefundPolicy.for(order)
result = RefundCalculator.call(order:, policy:)

执行顺序由代码确定，每一步都是写代码时就想好的。

Agent 场景里，程序可以向模型提供多个工具：

get_order
search_refund_policy
search_ticket
draft_reply

模型根据用户问题和当前上下文选择下一步动作，Python Runner 再把模型的选择映射到真实函数：

if tool_call.name == "get_order":
    result = get_order(**tool_call.arguments)
elif tool_call.name == "search_ticket":
    result = search_ticket(**tool_call.arguments)

看到这里我一下子不慌了，这有点像 Rails 里根据 next_action 分派不同的 Service Object。区别在于，Agent 里模型参与了语义判断和步骤选择，分派逻辑不再完全由我写死。

但这不意味着所有逻辑都交给模型。我的判断原则是：

语义理解和开放式选择交给模型，比如“用户的问题主要与订单、退款还是物流有关”。
权限、金额、状态转换和数据约束由程序负责，比如“退款金额超过 5000 元时必须由经理审批”，这是确定、可审计的规则。

一个最小的 Python Agent Loop

下面用 mock_model() 模拟模型决策。我想先把循环本身看清楚，所以故意不接真实模型。

from dataclasses import dataclass
from typing import Literal


ToolName = Literal["get_order"]


@dataclass
class ToolCall:
    name: ToolName
    arguments: dict[str, str]


@dataclass
class ModelMessage:
    kind: Literal["tool_call", "final"]
    content: str
    tool_call: ToolCall | None = None


def mock_model(messages: list[dict[str, str]]) -> ModelMessage:
    """模拟模型根据上下文决定调用工具或返回答案。"""
    has_tool_result = any(
        message["role"] == "tool"
        for message in messages
    )

    if has_tool_result:
        return ModelMessage(
            kind="final",
            content="订单已支付，建议回复客户：订单状态正常。",
        )

    return ModelMessage(
        kind="tool_call",
        content="需要先查询订单。",
        tool_call=ToolCall(
            name="get_order",
            arguments={"order_id": "ORDER-1001"},
        ),
    )


def get_order(order_id: str) -> dict[str, str]:
    """模拟订单数据库。"""
    orders = {
        "ORDER-1001": {
            "status": "paid",
            "customer": "Alice",
        }
    }

    if order_id not in orders:
        raise ValueError(f"order not found: {order_id}")

    return {
        "order_id": order_id,
        **orders[order_id],
    }


def run_agent(
    user_input: str,
    max_steps: int = 3,
) -> str:
    messages = [
        {
            "role": "user",
            "content": user_input,
        }
    ]

    for _ in range(max_steps):
        model_message = mock_model(messages)

        if model_message.kind == "final":
            return model_message.content

        if model_message.tool_call is None:
            raise RuntimeError("模型缺少工具调用信息")

        if model_message.tool_call.name == "get_order":
            result = get_order(
                **model_message.tool_call.arguments
            )

            messages.append(
                {
                    "role": "assistant",
                    "content": model_message.content,
                }
            )
            messages.append(
                {
                    "role": "tool",
                    "content": str(result),
                }
            )
            continue

        raise RuntimeError(
            f"unsupported tool: {model_message.tool_call.name}"
        )

    raise RuntimeError("agent exceeded max_steps")


answer = run_agent(
    "请帮我看看 ORDER-1001 怎么回复客户"
)

print(answer)

运行后得到：

订单已支付，建议回复客户：订单状态正常。

回看这段代码，每个部分的职责很清楚：

messages 保存模型当前能够看到的上下文。
mock_model() 模拟模型对下一步的判断。
get_order() 代表程序能够执行的真实工具。
run_agent() 负责控制整个执行过程。
max_steps 为循环设置硬终止条件。

注意 mock_model() 里没有“记忆”可言：它只看 messages 里有没有 tool 角色的消息，就决定是调工具还是给最终答案。真实的模型也一样，上下文就是它的全部世界。

顺手记下的几个 Python 概念

这几个概念是我今天查文档搞懂的，正好都出现在上面的 Agent 代码里，就对照着记在这里，免得以后忘。

`dataclass`

@dataclass
class ToolCall:
    name: ToolName
    arguments: dict[str, str]

dataclass 适合定义以保存数据为主要职责的类，可以类比 Ruby 的 Data.define，或者一个没有数据库持久化职责的 PORO。后面使用 Pydantic 时，还可以增加运行时数据校验。

`Literal`

Literal["tool_call", "final"]

表示字段只能取指定值。我一开始没写 Literal，直接用了 str，跑起来也没问题。后来想到一个问题：如果模型返回一个瞎编的工具名会怎样？这才明白类型声明是在替我把第一道关。Agent 系统里到处是有限状态（running、waiting、tool_call、completed……），写明确总比靠自觉好。

联合类型

tool_call: ToolCall | None = None

表示字段可能是 ToolCall，也可能是空。对应 Ruby 里的“对象或 nil”，只是 Python 把这种可能性明确写进了类型声明。

一个必须存在的安全边界

我最开始写的循环就是：

while True:
    response = call_model(messages)

跑通之后我才意识到一个问题：如果模型重复选择同一个工具，或者工具结果一直不满足任务，这个循环永远不会自己停下来。系统会持续消耗 token、时间和第三方 API 配额。

所以循环必须设置由程序控制的终止条件。最小实现用 for 就够了：

for _ in range(max_steps):
    ...

生产环境还需要补上：

最大执行步骤
最大 Token 数量
最大费用
最大执行时间
用户取消信号

这和 Rails 后台任务里的超时、重试次数及幂等控制很像。区别是 Agent 的执行路径更灵活，所以程序反而需要更严格地限制执行范围。

今天的 Review 检查项

跑通最小实现之后，我给自己列了一个清单，以后检查 AI 生成的 Agent 代码时先看这几条：

模型可以选择哪些工具？
工具参数是否经过校验？（就是 Literal 那道关）
权限判断发生在哪里？
工具异常会怎样传递和记录？
循环是否有明确的终止条件？（这条我今天真的踩过）
能否看到每次模型调用和工具调用的执行轨迹？

最后一项目前还没有。今天剩下的时间我准备给这个例子加上一份 trace：

trace: list[str] = []

让它输出：

model: tool_call
tool: get_order
model: final

我打算用这份轨迹验证一下自己是不是真的理解了最小 Agent Loop：模型做了什么决定，Python 执行了什么动作，任务为什么结束。能讲清楚这三件事，今天就算没白学。