AGENT

150 行 Python 构建生产级 AI 代理

工具调用、无状态状态管理和重试电路在 Anthropic Claude API 层的实际工作原理。

admin

Aug 30, 2026 • 9 min read

梯形图转SCL | AI模型价格对比 | AI工具导航 | ONNX模型库 | Vibe Coding教程 | PLC在线仿真器 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo

大多数开发者通过安装沉重的编排框架开始构建代理。一周之内，他们发现自己在调试多层抽象、追踪神秘的回调链，并与框架特定的提示注入作斗争。

你不需要一个 10,000 行的框架来运行 AI 代理。

其核心，代理只是一个包裹在无状态 API 调用周围的确定性 while 循环。直接在 Python 中构建这个循环可以让你完全控制工具执行、令牌预算、错误恢复和上下文管理。

以下是使用 Anthropic 的 Python SDK 和现代 Claude 模型在底层实际工作的机制。

1、LLM API 的无状态现实

像 Claude 或 ChatGPT 这样的 Web 界面感觉是有状态的，因为浏览器为你维护对话状态。底层 API 则完全不是这样。

在核心银行系统中，转账会更新关系数据库中的账户余额。服务器存储该状态，因为余额记录只有几个字节。LLM 推理端点无法做到这一点。由于多轮对话在数百万并发用户中消耗数千个令牌，Anthropic 保持推理端点完全无状态。

每次 API 调用都是隔离的。如果你需要多轮记忆，你的客户端必须在每个请求上发送完整的消息历史。

import anthropic
client = anthropic.Anthropic()
# 无状态多轮对话：客户端每轮发送完整历史
messages = [
    {"role": "user", "content": "法国的首都是什么？"},
    {"role": "assistant", "content": "法国的首都是巴黎。"},
    {"role": "user", "content": "它的人口是多少？"}
]
response = client.messages.create(
    model="claude-5-sonnet",
    max_tokens=300,
    messages=messages
)

当你需要模型执行操作时，这个线性历史会转变为执行循环。

2、决策-行动-观察循环

代理将基本的 LLM 推理扩展为自主决策循环。

提示和工具架构： 客户端用精确的 JSON 架构定义可用工具并传递给 Claude。
模型决策： Claude 分析请求。如果需要外部数据或操作，它会停止文本生成并设置 stop_reason == "tool_use"。
本地执行： 你的应用程序检查工具调用，本地执行匹配的 Python 函数，并捕获结果。
观察反馈： 客户端将结果格式化为 tool_result 块并附加到消息历史。
重新评估： Claude 读取新上下文，然后调用另一个工具或以 stop_reason == "end_turn" 完成任务。

[用户提示] -> [Claude API]
                      |
           (stop_reason == "tool_use")
                      |
                      v
             [执行本地工具]
                      |
             [附加 tool_result]
                      |
                      v
         [Claude API（下一轮）] -> (stop_reason == "end_turn") -> [最终输出]

实现核心循环

这是一个使用 'claude-5-sonnet' 的完整、最小代理循环

import anthropic
client = anthropic.Anthropic()

# 1. 用清晰的架构和描述定义工具
tools = [
    {
        "name": "search_database",
        "description": "搜索内部产品目录的库存、定价和库存状态。",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "产品搜索词"},
                "category": {
                    "type": "string",
                    "enum": ["hardware", "software", "services"],
                    "description": "可选类别过滤器"
                }
            },
            "required": ["query"]
        }
    }
]

def search_database(query: str, category: str = "hardware") -> str:
    # 模拟数据库查找
    return f"找到 1 个项目：'{query}' 在 {category} 中（SKU：HW-884）- 价格：$499，有库存：是"

# 2. 执行循环
messages = [{"role": "user", "content": "检查我们是否有 Pro 笔记本电脑库存。"}]
max_iterations = 10

for step in range(max_iterations):
    response = client.messages.create(
        model="claude-5-sonnet",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    # 将助手的响应附加到历史
    messages.append({"role": "assistant", "content": response.content})

    if response.stop_reason == "tool_use":
        for block in response.content:
            if block.type == "tool_use":
                if block.name == "search_database":
                    result = search_database(**block.input)
                else:
                    result = f"错误：未知工具 '{block.name}'"

                # 将工具输出反馈给模型
                messages.append({
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        }
                    ]
                })
    elif response.stop_reason == "end_turn":
        for block in response.content:
            if hasattr(block, "text"):
                print("最终答案：", block.text)
        break

注意 tool_use_id 是如何在 tool_result 块中传回的。Claude 使用此 ID 将每个执行结果映射回其原始意图。

3、工具设计：避免架构膨胀

你定义工具的方式直接决定了模型的准确性。

精确的文档字符串： Claude 严重依赖描述字符串来选择工具。如果你有相似的函数（search_web、search_internal_docs、search_customer_db），在每个描述中写明不同的边界条件。
参数化胜于函数蔓延： 不要为搜索不同目标构建三个独立函数。构建一个带有 target 枚举（web、docs、database）的 search 工具。这节省了架构令牌并防止工具选择混淆。
系统级回退： 添加系统提示，指示模型在查询无法满足时声明，而不是编造工具参数。

4、在不崩溃循环的情况下处理故障

在生产环境中，循环在三个不同层面上失败：

API 基础设施： HTTP 429 速率限制、HTTP 500/529 服务器过载、网络中断。
工具执行： 数据库超时、无效输入类型、下游 API 中断。
模型错误： 幻觉工具名称或架构违规。

你的循环必须将错误分类为可重试和不可重试类别。

带抖动的指数退避

速率限制和 529 过载等瞬态错误应自动重试。身份验证错误（401）或格式错误的请求（400）应立即失败。

import time
import anthropic

def execute_with_retry(client, messages, tools, max_retries=3):
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model="claude-5-sonnet",
                max_tokens=1024,
                tools=tools,
                messages=messages
            )
        except anthropic.RateLimitError as e:
            if attempt == max_retries - 1:
                raise e
            sleep_time = base_delay * (2 ** attempt)
            time.sleep(sleep_time)
        except anthropic.APIStatusError as e:
            # 500 和 529 是瞬态过载代码
            if e.status_code in [500, 529] and attempt < max_retries - 1:
                sleep_time = base_delay * (2 ** attempt)
                time.sleep(sleep_time)
            else:
                raise e

优雅的模型回退

如果你的主模型（claude-5-sonnet）遇到持续的速率限制或延迟增加，你的循环可以回退到 claude-4–5-haiku 以保持正常运行时间。

def call_claude_with_fallback(client, messages, tools):
    primary_model = "claude-5-sonnet"
    fallback_model = "claude-4-5-haiku"

    try:
        return client.messages.create(
            model=primary_model,
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
    except anthropic.APIError:
        # 回退到高速、经济高效的模型
        return client.messages.create(
            model=fallback_model,
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

工具级断路器

当本地工具崩溃时，永远不要抛出会终止进程的未捕获异常。捕获异常并在 tool_result 块内返回结构化错误消息。

def safe_tool_execution(tool_name: str, tool_args: dict) -> str:
    try:
        if tool_name == "search_database":
            return search_database(**tool_args)
        return f"错误：未识别工具 '{tool_name}'。"
    except TypeError as e:
        return f"工具参数错误：缺少或无效参数（{str(e)}）。请更正你的参数。"
    except Exception as e:
        return f"工具执行失败：{type(e).__name__}（{str(e)}）。考虑替代方法。"

通过将错误返回给 Claude，模型可以推理失败、调整参数或向用户解释限制。

5、上下文卫生：序列化优于急切转储

代理设计中的一个常见错误是急切上下文转储：将用户配置文件、完整文档和数据库架构塞入系统提示。

这会增加提示令牌成本并降低检索精度。相反，使用自适应上下文收集：为代理提供有针对性的检索工具，使其仅获取当前步骤所需的确切数据切片。

返回工具输出时，在返回前清理数据。永远不要将原始 JSON 转储、具有空列的数据库行或原始 HTML 发送到 tool_result。

def format_user_profile_for_llm(raw_db_row: dict) -> str:
    # 仅将可操作字段提取为紧凑文本
    return (
        f"用户 ID：{raw_db_row.get('id')}\n"
        f"层级：{raw_db_row.get('subscription_tier')}\n"
        f"区域：{raw_db_row.get('region')}\n"
        f"活跃项目：{', '.join(raw_db_row.get('projects', []))}"
    )

将 2KB 嵌套 JSON 块转换为四行纯文本，每轮可节省数百个令牌。在 10 轮代理运行中，这会显著累积。

6、确定性边界和终止保护

没有确定性边界的代理最终会遇到无限循环。当模型重复调用失败工具或在两个操作之间循环时，就会发生这种情况。

每个生产循环都需要两个硬限制：迭代限制和累积令牌预算。

MAX_TURNS = 10
TOTAL_TOKEN_BUDGET = 50000
accumulated_tokens = 0

for turn in range(1, MAX_TURNS + 1):
    response = client.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    # 跟踪输入和输出的使用量
    accumulated_tokens += (response.usage.input_tokens + response.usage.output_tokens)

    if accumulated_tokens > TOTAL_TOKEN_BUDGET:
        print(f"已终止：超出 {TOTAL_TOKEN_BUDGET} 个令牌的令牌预算。")
        break

    if response.stop_reason == "end_turn":
        print(f"目标在 {turn} 轮内成功完成。")
        break

    if turn == MAX_TURNS:
        print("已终止：达到最大轮次上限。返回最佳努力结果。")

如果你需要基于质量的停止，请运行一个辅助评估步骤，根据验收标准验证输出。如果连续轮次的质量分数增量为零，则提前退出循环。

7、工具边界上的防御性验证

LLM 可能会产生幻觉参数、翻转日期或忽略单位。在将工具输出馈送到后续操作时，永远不要盲目信任。

检测矛盾： 如果用户指定"上午 9:00 之前出发"，而航班预订工具返回下午 2:00 的时段，请提示模型标记不匹配，而不是继续预订。
应用平衡验证：对于支付或数据删除等高后果操作，使用严格的 Pydantic 架构。对于网络搜索和文本摘要等信息查询，保持验证宽松，以避免因不必要的重试而浪费令牌。

8、核心规则

框架使原型设计变得快速，但它们隐藏了使用 LLM 构建的实际运营现实。

当你自己编写代理循环时，你可以控制通过网络发送的每个令牌、每个重试延迟和每个终止边界。

从 150 行干净的 Python 开始。仅在你的领域逻辑真正需要时才添加抽象。

原文链接：How to Build a Production AI Agent in 150 Lines of Python

汇智网翻译整理，转载请标明出处