第一个 Agent：从零手写

哓番茄大约 4 分钟

第一个 Agent：从零手写

不装任何框架，用原生 SDK 从零写出一个带工具、带流式的 agent。写完这一篇，你对「agent 到底是什么」的疑问会全部消失。

一、地基：messages 是状态机

agent 开发的第一课：对话历史（messages）就是你应用的状态。

messages = [
    {"role": "user", "content": "你好"},
]
resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=messages,
)

messages 的三个角色构成一个不断追加的数组：

user      →  用户输入 / 工具执行结果
assistant →  模型回复（文本 / 工具调用请求）
system    →  系统提示（不在 messages 里，单独传，永不被截断）

规则：每次调用把完整历史发回去（无状态 API），
     模型的「记忆」全靠你维护这个数组

多轮对话的最小实现：

messages = []

while True:
    user_input = input("你: ")
    if user_input in ("quit", "exit"):
        break

    messages.append({"role": "user", "content": user_input})
    resp = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=messages,
    )
    text = "".join(b.text for b in resp.content if b.type == "text")
    print(f"AI: {text}")

    # 关键：把模型回复追加回历史，否则模型失忆
    messages.append({"role": "assistant", "content": resp.content})

为什么必须回填 assistant 消息？API 是无状态的；少 append 这一步，下一轮模型看不到自己说过什么——多轮一致性全毁。

二、加工具：从聊天到干活

工具（function calling）是 agent 和 chatbot 的分水岭。流程：

你注册工具 schema → 模型决定调用 → 返回 tool_use 块
     ↓
你的代码执行真正动作 → 把结果以 tool_result 回注
     ↓
模型看到结果，继续推理（再调工具 or 给最终答案）

完整实现：

import json
from anthropic import Anthropic

client = Anthropic()

TOOLS = [{
    "name": "get_weather",
    "description": "查询指定城市当前天气。当用户询问天气相关问题时使用。",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "城市名，如「北京」"},
        },
        "required": ["city"],
    },
}]

def get_weather(city: str) -> str:
    # 真实项目里这里是 HTTP 调用；用假数据聚焦流程
    return json.dumps({"city": city, "temp": "26°C", "cond": "晴"}, ensure_ascii=False)

HANDLERS = {"get_weather": get_weather}

SYSTEM = "你是天气助手。涉及天气的问题必须先调用工具查询，不要编造。"

def agent(task: str, max_turns: int = 10) -> str:
    messages = [{"role": "user", "content": task}]

    for _ in range(max_turns):  # 轮数上限 = 失控刹车
        resp = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1024,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
        )

        # 执行本轮全部工具调用
        tool_results = []
        for block in resp.content:
            if block.type == "tool_use":
                result = HANDLERS[block.name](**block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

        if not tool_results:          # 没有工具调用 = 任务完成
            return "".join(b.text for b in resp.content if b.type == "text")

        messages.append({"role": "assistant", "content": resp.content})
        messages.append({"role": "user", "content": tool_results})

    return "（超过最大轮数，强制停止）"

print(agent("北京今天适合跑步吗？"))

这 50 行就是所有 agent 的骨架——LangGraph、CrewAI 内部都是这个循环。运行时你会看到模型先 tool_use 拿到天气，再结合温度给出「适合跑步」的判断。

循环每个环节的运行时原理（错误当数据、刹车机制等）在 Harness 核心循环篇有更深的展开。

三、流式输出：体验的分水岭

非流式：用户盯着空白 20 秒。流式：字一点点出来，工具调用实时可见。

import sys

def agent_stream(task: str, max_turns: int = 10):
    messages = [{"role": "user", "content": task}]

    for _ in range(max_turns):
        with client.messages.stream(
            model="claude-sonnet-5",
            max_tokens=1024,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
        ) as stream:
            for event in stream:
                if event.type == "content_block_delta":
                    if event.delta.type == "text_delta":
                        sys.stdout.write(event.delta.text)
                        sys.stdout.flush()
            final = stream.get_final_message()   # 拿完整消息做状态回填

        tool_results = []
        for block in final.content:
            if block.type == "tool_use":
                print(f"\n[调用工具] {block.name}({block.input})")
                result = HANDLERS[block.name](**block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

        if not tool_results:
            print()          # 流式文本已边收边打
            return

        messages.append({"role": "assistant", "content": final.content})
        messages.append({"role": "user", "content": tool_results})

流式事件的类型序列：

message_start → content_block_start(text/tool_use)
→ content_block_delta(...)*    ← 文本增量在这，UI 渲染挂这里
→ content_block_stop → message_delta(usage) → message_stop

四、健壮性：重试与超时

生产代码里裸调 API 必挂。第一天就包一层：

import time
from anthropic import APIStatusError, APITimeoutError, RateLimitError

def call_llm(**kwargs) -> object:
    for attempt in range(3):
        try:
            return client.messages.create(timeout=60, **kwargs)
        except RateLimitError:            # 429：退避重试
            time.sleep(2 ** attempt)
        except APITimeoutError:
            if attempt == 2:
                raise
        except APIStatusError as e:       # 5xx 可重试，4xx 不要
            if 500 <= e.status_code < 600:
                time.sleep(1)
            else:
                raise
    raise RuntimeError("LLM 调用重试耗尽")

要点：

□ 只重试「可重试错误」：429 / 5xx / 超时；400（参数错）重试没意义
□ 指数退避：1s → 2s → 4s，加随机抖动防止同步风暴
□ timeout 必须显式设：SDK 默认值可能长达 10 分钟
□ SDK 的 stream + tool 并发调用时，注意重试会重复计费（打点要准）

五、TS 对照版（15 行骨架）

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

export async function agent(task: string, tools: any[], handlers: Record<string, Function>, maxTurns = 10) {
  const messages: any[] = [{ role: "user", content: task }];

  for (let i = 0; i < maxTurns; i++) {
    const resp = await client.messages.create({
      model: "claude-sonnet-5", max_tokens: 1024, messages, tools,
    });

    const results = resp.content
      .filter((b: any) => b.type === "tool_use")
      .map((b: any) => ({
        type: "tool_result" as const,
        tool_use_id: b.id,
        content: String(handlers[b.name](b.input)),
      }));

    if (!results.length)
      return resp.content.filter((b: any) => b.type === "text").map((b: any) => b.text).join("");

    messages.push({ role: "assistant", content: resp.content });
    messages.push({ role: "user", content: results });
  }
  return "（超过最大轮数）";
}

本篇小结

messages 是 agent 的状态机：assistant 回填一步都不能少
50 行循环 = 一切 agent 的骨架：调模型 → 执行工具 → 回注 → 直到无工具调用
流式输出做进 MVP，不要留到「以后优化」
重试/超时/轮数上限是第一天的事，不是上线前的事

下一篇：工具开发实战——真实世界的工具怎么设计、怎么防错、怎么测试。