实操：从零写一个最小 Agent

一句话理解：一个能用的 Agent Loop 只需要四样东西：工具定义、一个“调模型 → 执行工具 → 回填结果”的循环、一个停止条件，以及把工具错误交还给模型而不是让程序崩溃的处理；百行代码就能写完。

前面几课讲了 Harness 的全景、组件、Hooks、错误恢复和框架选型。如果你从没亲手写过循环，这些概念很容易停留在“听懂了”的层面。这一课直接动手：不用任何 Agent 框架，只用 Anthropic 官方 Python SDK 调用 Claude API，写一个「订单售后诊断 Agent」。

写完之后，你会对三件事有体感：模型到底输出了什么；“工具调用”在代码里长什么样；框架替你做掉的那部分工作究竟有多少。

我们要做什么

用户问“订单 ORD1001 的退款为什么还没到账？”，Agent 需要：

1. 调用 get_order 查订单状态和承诺退款日期；
2. 调用 get_payment 查支付与退款记录；
3. 根据两份数据给出结论、依据和下一步建议。

为了让你不依赖任何外部系统就能跑通，两个工具读取的是代码里的模拟数据。把它们换成真实接口，就是课程实战项目「订单售后诊断 Agent」的雏形。

整个循环的流程如下：

准备环境

• Python 3.10 或更高版本（Anthropic Python SDK 1.x 的要求）。
• 一个 Claude API Key，在 Claude Console 中创建。

mkdir mini-agent && cd mini-agent
python3 -m venv .venv
source .venv/bin/activate        # Windows 使用 .venv\Scripts\activate
pip install anthropic
export ANTHROPIC_API_KEY="你的 API Key"

完整代码

新建 agent.py，粘贴以下代码。代码已对照 Claude API 的工具调用文档编写，并在 anthropic 1.6.0 下用模拟响应验证过循环、错误处理与步数上限逻辑。

"""最小 Agent Loop：订单售后诊断 Agent（两个只读工具 + 最大步数 + 工具错误处理）"""
import json
import os
import sys
import time

import anthropic

MODEL = os.environ.get("AGENT_MODEL", "claude-opus-5")
MAX_STEPS = 6  # 最多调用模型的次数；到达上限就停下，防止死循环和成本失控

# ---------- 1. 模拟两个业务系统的数据 ----------
ORDERS = {
    "ORD1001": {"status": "已退货入库", "amount": 428.0, "created_at": "2026-09-13", "refund_due_date": "2026-09-15"},
    "ORD2002": {"status": "已完成", "amount": 129.0, "created_at": "2026-09-10", "refund_due_date": "无"},
}
PAYMENTS = {
    "ORD1001": {
        "paid_at": "2026-09-13 08:10", "amount": 428.0,
        "refund_status": "退款审核中", "refund_updated_at": "2026-09-14 09:30",
        "refund_note": "收款账户信息异常，待联系用户确认",
    },
}


# ---------- 2. 工具实现：出错就抛出“可行动”的错误信息 ----------
def get_order(order_id: str) -> dict:
    if order_id not in ORDERS:
        raise ValueError(f"订单 {order_id} 不存在，请让用户核对订单号（示例：ORD1001）")
    return ORDERS[order_id]


def get_payment(order_id: str) -> dict:
    if order_id not in PAYMENTS:
        raise ValueError(f"支付系统暂无 {order_id} 的支付记录，可能为货到付款或数据未同步")
    return PAYMENTS[order_id]


TOOL_IMPLS = {"get_order": get_order, "get_payment": get_payment}

ORDER_ID_SCHEMA = {
    "type": "object",
    "properties": {"order_id": {"type": "string", "description": "订单号，例如 ORD1001"}},
    "required": ["order_id"],
}
TOOLS = [
    {
        "name": "get_order",
        "description": "查询订单基本信息：状态、金额、下单日期、承诺退款日期。用户给出订单号时先调用它。",
        "input_schema": ORDER_ID_SCHEMA,
    },
    {
        "name": "get_payment",
        "description": "查询订单的支付与退款记录（支付时间、金额、退款状态与备注）。需要判断退款延迟或异常原因时调用。",
        "input_schema": ORDER_ID_SCHEMA,
    },
]

SYSTEM = (
    "你是订单售后诊断助手。只依据工具返回的数据下结论，不要编造。"
    "工具报错时，根据错误信息决定换个方式查询或如实告知用户。"
    "最终回答用三行：结论、依据、建议的下一步。"
)


def run_tool(name: str, args: dict) -> tuple[str, bool]:
    """执行一个工具，返回 (结果文本, 是否出错)。异常不让程序崩溃，而是交还给模型。"""
    impl = TOOL_IMPLS.get(name)
    if impl is None:
        return f"未知工具 {name}，可用工具：{list(TOOL_IMPLS)}", True
    try:
        return json.dumps(impl(**args), ensure_ascii=False), False
    except TypeError as e:  # 参数缺失或多余
        return f"参数错误：{e}", True
    except Exception as e:  # 业务错误、超时等
        return f"{type(e).__name__}: {e}", True


def run_agent(task: str) -> str:
    client = anthropic.Anthropic()  # 从环境变量 ANTHROPIC_API_KEY 读取凭据
    messages = [{"role": "user", "content": task}]

    for step in range(1, MAX_STEPS + 1):
        response = client.messages.create(
            model=MODEL, max_tokens=16000, system=SYSTEM, tools=TOOLS, messages=messages
        )

        if response.stop_reason == "end_turn":  # 模型不再调用工具：任务结束
            return "".join(b.text for b in response.content if b.type == "text")

        if response.stop_reason != "tool_use":  # max_tokens、refusal 等：不假装成功
            return f"[提前停止] stop_reason={response.stop_reason}"

        # 必须把完整的 content（含 tool_use 块）原样放回历史
        messages.append({"role": "assistant", "content": response.content})

        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            start = time.perf_counter()
            output, is_error = run_tool(block.name, block.input)
            ms = (time.perf_counter() - start) * 1000
            flag = "ERROR" if is_error else "ok"
            print(f"[step {step}] {block.name}({json.dumps(block.input, ensure_ascii=False)}) -> {flag} {ms:.1f}ms")
            results.append(
                {"type": "tool_result", "tool_use_id": block.id, "content": output, "is_error": is_error}
            )

        # 同一轮的所有工具结果放进同一条 user 消息
        messages.append({"role": "user", "content": results})

    return f"[提前停止] 已达到最大步数 {MAX_STEPS}，任务未完成"


if __name__ == "__main__":
    question = sys.argv[1] if len(sys.argv) > 1 else "订单 ORD1001 的退款为什么还没到账？"
    try:
        print(run_agent(question))
    except anthropic.AuthenticationError:
        sys.exit("认证失败：请检查 ANTHROPIC_API_KEY")
    except anthropic.APIConnectionError:
        sys.exit("网络连接失败：请检查网络或代理设置")
    except anthropic.APIStatusError as e:
        sys.exit(f"API 返回错误 {e.status_code}：{e.message}")

逐段理解

先看全局：agent.py 的六段代码，各自对应 Harness 的一项职责。

工具定义（TOOLS）：模型只能通过名称、描述和参数 Schema 理解工具。描述里写清“什么时候该调用”，比只写“做什么”更能帮助模型做出正确选择。详见 工具定义三要素。

循环主体：每一轮先调用模型，再看 stop_reason：

stop_reason	含义	代码的处理
tool_use	模型想调用工具	执行工具，回填结果，进入下一轮
end_turn	模型认为任务完成	拼出文本回答，结束
max_tokens	输出被长度上限截断	如实报告，不当作成功
refusal	模型出于安全原因拒绝	如实报告，不当作成功

两条容易写错的格式规则（来自 Claude API 文档）：

1. 模型的回复要 原样 追加到历史中，包括 tool_use 块（以及可能出现的思考块）。只拼接文字会导致下一次请求报错或丢失上下文。
2. 工具结果必须紧跟在对应的工具调用之后，放在一条 user 消息中，并用 tool_use_id 与调用一一对应。模型一次发起多个工具调用时，所有结果放进同一条消息。

以模型同一轮并行调用两个工具为例，消息历史长这样：

错误处理（run_tool）：三类错误都被转成 is_error: True 的工具结果，而不是抛出异常中断程序：未知工具、参数错误、业务错误。模型收到错误信息后，可以修正参数、换一种查法，或如实告诉用户。这正是 错误恢复与重试 中“把错误翻译给模型”的最小实现。

停止条件（MAX_STEPS）：这里的“步”指调用模型的次数。上限是防止死循环和成本失控的最后一道闸门，更完整的策略见 停止条件。

模型 API 层的错误：限流、服务端错误和网络问题会先由 SDK 自动重试（默认两次）；仍失败时抛出异常，由 __main__ 中的 except 给出可读提示。

运行与预期输出

python agent.py

你会看到类似下面的输出。工具调用顺序可能不同（模型也可能在同一轮并行调用两个工具），最终回答的措辞每次都会有差异：

[step 1] get_order({"order_id": "ORD1001"}) -> ok 0.0ms
[step 2] get_payment({"order_id": "ORD1001"}) -> ok 0.0ms
结论：订单 ORD1001 的退款仍卡在审核环节，尚未打款（承诺退款日期为 2026-09-15）。
依据：支付记录显示 9 月 14 日 09:30 退款状态仍为“退款审核中”，备注为“收款账户信息异常，待联系用户确认”。
建议的下一步：联系用户核实并更正收款账户，然后重新提交这笔 428 元的退款审核。

再试两个会触发错误处理的问题：

python agent.py "帮我看下订单 ORD2002 的支付和退款记录"
python agent.py "订单 ORD3003 的退款到哪一步了？"

第一个问题中，get_payment 会返回错误（模拟数据里没有 ORD2002 的支付记录），你应该看到 -> ERROR，而模型会基于订单状态“已完成”给出回答，并说明支付与退款数据暂缺。第二个问题中，订单不存在，模型应该请你核对订单号，而不是编造一个退款进度。

排错提示

现象	可能原因	处理
ModuleNotFoundError: No module named 'anthropic'	没有在虚拟环境中安装 SDK	激活虚拟环境后重新 pip install anthropic
TypeError: Could not resolve authentication method	没有设置 API Key	在当前终端执行 export ANTHROPIC_API_KEY=...
输出“认证失败”	API Key 错误、被撤销或过期	到 Claude Console 重新生成
API 返回错误 400 且提到 tool_use 与 tool_result	历史消息格式不对	确认追加的是 response.content 原对象；工具结果紧跟其后且 ID 对应
API 返回错误 404 或提示模型不存在	模型名拼写错误或账号无权使用	检查 AGENT_MODEL 环境变量，使用 Claude 文档中列出的模型 ID
API 返回错误 429	触发限流或花费上限	稍后重试；检查 Console 中的用量与额度
输出“已达到最大步数”	模型在反复调用工具	查看打印的调用记录，检查工具描述是否含糊、错误信息是否可行动
模型不调用工具、直接编造答案	工具描述没写清何时调用，或指令约束不足	在描述中写明触发条件，在系统指令中强调“只依据工具数据”

默认模型是 claude-opus-5。如需换成其他 Claude 模型，设置环境变量即可，例如 export AGENT_MODEL=claude-sonnet-5，不用改代码。

从最小循环到生产 Harness，还差什么

对照 Harness 的组件，这一百行代码只覆盖了循环控制、工具执行和最基础的错误处理。还缺：

缺少的能力	为什么需要	相关课程
权限与审批	写操作不能自动执行	Human-in-the-loop
工具层重试与超时	真实接口会抖动	错误恢复与重试
会话持久化	进程退出后对话就丢了	会话与持久化
上下文管理	长对话会撑满窗口	长任务的上下文管理
追踪与成本统计	事后要能回放和算账	Tracing
结果验证	模型说完成不代表真的完成	为什么必须验证

生产中如果不想手写循环，Anthropic SDK 提供 Beta 的 Tool Runner：你只定义工具函数，循环由 SDK 驱动，同时保留每一轮介入的能力。不过先手写一次，你才看得懂 Tool Runner 和各类框架替你做了什么。

常见误区

• “只把模型回复里的文字存进历史就够了。” 丢掉 tool_use 块，下一轮的 tool_result 就找不到对应的调用，请求会直接报错。
• “工具出错就抛异常，让外层兜底。” 这样模型永远没有机会修正参数或换条路径，一个小错误就会终结整个任务。
• “没设步数上限也没关系，模型自己会停。” 大多数时候会，但你需要为少数不会的情况负责，尤其是涉及费用的时候。

练习

在上面的代码基础上完成以下改造，每项 10–20 分钟：

1. 记录轨迹：把每次工具调用写入 trace.jsonl，每行包含步数、工具名、参数、是否出错、耗时。
2. 加第三个工具：get_after_sales_tickets(order_id)，返回该订单的历史售后工单；观察模型何时会调用它。
3. 打转检测：如果同一工具以相同参数被调用两次，在第二次的工具结果中提醒模型“结果不会变化”。
4. 统计花费：累加每次响应中 response.usage 的输入与输出 Token，运行结束时打印总量。

延伸阅读

• Handle tool calls：tool_use 与 tool_result 的格式规则、is_error 的用法。
• Claude API errors：错误码与 SDK 自动重试行为。
• Building effective agents：为什么建议先直接用 API。

下一步

这个最小 Agent 还缺很多 Harness 该做的事：上下文装配、权限把关、Hooks 和运行日志都没有。下一课「实操：把最小 Agent 升级成迷你 Harness」就在这份 agent.py 的基础上，把本模块讲过的机制一项项补上。