部分内容由豆包辅助生成

一个 AI Agent，大概率会从"调通一个 LLM API"开始。这件事看起来简单——发个 HTTP POST，拿个 JSON 回来，完事。

但真正把它做成一个能跑多轮对话、能接工具调用、能流式输出的 Agent 基础设施时，你会发现里面藏着不少工程上的坑。

本文以 Claude 的 Messages API 为主线，结合一个终端 Coding Agent的实际设计，把 LLM 客户端和对话管理器这两层基础设施讲透。

一、Messages API

先看一个最基础的请求。用 curl 直接打 Anthropic 的 Messages API：

curl https://api.anthropic.com/v1/messages \
    -H 'Content-Type: application/json' \
    -H 'anthropic-version: 2023-06-01' \
    -H "X-Api-Key: $ANTHROPIC_API_KEY" \
    -d '{
          "max_tokens": 1024,
          "messages": [
            {
              "content": "Hello, world",
              "role": "user"
            }
          ],
          "model": "claude-sonnet-4-6"
        }'

响应（省略了部分字段）长这样：

{
  "id": "msg_013Zva2CMHLNnXjNJJKqJ2EF",
  "type": "message",
  "role": "assistant",
  "content": [
    {
      "type": "text",
      "text": "Hello! How can I help you today?"
    }
  ],
  "model": "claude-sonnet-4-6",
  "stop_reason": "end_turn",
  "usage": {
    "input_tokens": 10,
    "output_tokens": 12
  }
}

一个 HTTP POST，一段 JSON 过去，一段 JSON 回来。所有 LLM 应用本质上都在干这件事，只是上面包了不同的壳。但这个简单的 JSON 里有两个值得琢磨的细节。

1.1 user 和 assistant 必须交替

请求里的 messages 是一个数组，每条消息有 role 和 content 两个字段。role 只有两个值：user（用户）和 assistant（助手）。

Claude 是按 user/assistant 交替对话的模式训练的，所以 messages 数组里最好保持两个角色交替出现，第一条通常是 user。

API 不会因为连续两条 user 就报错——它会自动合并。但连续两条 assistant 会直接被拒绝：

{
  "type": "error",
  "error": {
    "type": "invalid_request_error",
    "message": "messages: roles must alternate between \"user\" and \"assistant\", but found multiple \"assistant\" roles in a row"
  }
}

这个坑在实现工具调用时一定会踩到。LLM 返回一个工具调用请求，这算 assistant 消息。

你执行完工具拿到结果，需要把结果作为 user 消息发回去——因为从 API 视角看，所有你发给模型的东西都归 user。如果搞错了角色，把工具结果也当 assistant 发出去，就会触发上面的报错。

1.2 响应的 content 永远是数组

注意响应里的 content 字段——它永远是一个数组，不是字符串。

虽然发送请求时 content 可以用字符串简写，但响应里永远是数组格式。

为什么是数组？因为模型的一次回复可能包含多种内容：先说一段话，然后请求调用一个工具，甚至一次调多个工具。

每种内容是一个独立的 content block，类型可能是 text、tool_use、thinking 等。

现在你只会看到 type: "text" 的内容块。但到实现工具系统时你就会遇到 tool_use。

如果忘了 content 是数组，代码大概率会出 bug。

二、流式响应

前面的 curl 示例用的是普通请求：模型生成完所有内容，一次性返回。

调试时没问题，产品里完全不能用。

原因很简单：Claude 生成一段长回复可能需要 10 到 30 秒。用户盯着空白屏幕等 30 秒？用户体验不太友好，所以必须用流式响应。

模型一边生成一边推送，你一边收到一边显示，用户看到字一个一个蹦出来，就像有人在实时打字。

2.1 SSE 事件序列

流式响应基于 SSE（Server-Sent Events）协议，本质是一个长连接的 HTTP 响应，服务器往里面持续写数据。Claude 的流式事件有固定顺序：

message_start          整个响应开始，带着 input_tokens 信息
  └─ content_block_start   一个内容块开始（文本或工具调用）
       └─ content_block_delta  内容增量，文字一个词一个词地到达
  └─ content_block_stop    一个内容块结束
message_delta            消息级别的增量（output_tokens、停止原因）
message_stop             整个响应结束

你的代码需要在不同事件上做不同的事：

• message_start 到达时记录输入 Token 数

• content_block_delta 每来一个就把文字增量推给 UI 显示

• message_delta 里提取输出 Token 数和停止原因

• message_stop 做收尾

容易踩的坑：一次响应可能有多个 content_block。比如模型先输出一段文字，再请求调用一个工具，这就是两个 block。解析代码不能假设只有一个文本块，要留好扩展空间。

2.2 不同语言的流式处理模式

流式处理的核心需求是一样的：生产者持续产生事件，消费者逐个处理。但不同语言有不同的惯用模式：

不管哪种，核心就是让生产者和消费者各干各的、互不阻塞。流跑完了连接要自动关掉，不能漏资源。

用户按 Ctrl+C 的时候得能干净退出，不能挂在那儿。

三、一个请求里的三个抽屉

Claude API 的请求里有三个不同的信息字段：

把三个字段放在一起，一个完整的 API 请求长这样：

{
  "model": "claude-sonnet-4-6",
  "max_tokens": 4096,
  "system": "你是 SxuanCode，一个终端环境中的 AI 编程助手。\n\n# Environment\n当前工作目录: /home/dev/myproject\n操作系统: Linux\n当前时间: 2026-05-27",
  "messages": [
    {"role": "user", "content": "帮我读一下 app.py 的内容"},
    {"role": "assistant", "content": "好的，我来读取 app.py 的内容。\ndef main():\n    print(\"hello\")\n\nif __name__ == \"__main__\":\n    main()"},
    {"role": "user", "content": "这个文件里有什么函数？"}
  ],
  "tools": [
    {
      "name": "read_file",
      "description": "读取指定路径的文件内容",
      "input_schema": {
        "type": "object",
        "properties": {
          "path": {"type": "string", "description": "文件路径"}
        },
        "required": ["path"]
      }
    }
  ]
}

system 里放角色设定和环境信息——你总不希望模型在 Linux 上给你建议用 PowerShell 吧。

messages 是正常的对话历史，user 和 assistant 交替出现。tools 里声明 Agent 能用的工具，这就是 Function Calling 的入口。

四、Token

Token 是 LLM 的计费单位，一般各家都会有自家的分词器Tokenizer，而且国外的比如Anthropic还会对中文征收中文税（同一个Prompt，用中文会更耗费Token）。

一般来说：英文每个单词大约 1-2 个 token，中文每个字大约 1-2 个 token。

具体取决于模型的 tokenizer，不需要精确计算，只需要知道它是衡量输入输出量和计费的基本单位。

回头看响应里的 usage 字段：

"usage": {
  "input_tokens": 10,
  "output_tokens": 12
}

Claude API 的计费分两部分：input_tokens 是你发给模型的所有内容（包括 system prompt、messages 和 tools 描述），output_tokens 是模型生成的回复。

• 输出 Token 比输入 Token 贵得多，缓存命中的计费倍率也能很大程度影响计费。

但更隐蔽的成本陷阱在多轮对话。

每一轮请求，你都要把完整的对话历史发过去。聊了 20 轮，第 21 轮请求就包含前 20 轮的所有消息，input_tokens 随对话轮次线性增长。

算一笔账：假设每次请求的固定开销（system prompt、环境信息、工具描述等）是 1000 tokens，每轮用户输入 50 tokens，模型回复 500 tokens。到第 20 轮，仅 input_tokens 就是 1000 + 20 × (50 + 500) = 12000 tokens。而第 1 轮只有 1050 tokens，差了 10 倍还多。

这就是为什么要上下文压缩。

五、Extended Thinking：让模型先想再说

Claude 支持 Extended Thinking，让模型在正式回复之前先进行一轮内部推理。开启后响应的 content 数组里会多一个 thinking 类型的内容块，排在 text 块之前。

对 Agent 开发来说，记住两件事：

1. thinking 的 Token 算在 output_tokens 里，是有成本的，本质上是用钱换更准确的工具调用决策。Agent 场景下通常值得——一次准确的工具调用可以省掉好几轮纠错的开销。

2. thinking 内容不能放进后续请求的 messages 里。维护对话历史时必须把 thinking 块过滤掉，只保留 text 和 tool_use 块发给 API，否则 API 会报错。

六、封装的核心原则：暴露领域语义，隐藏实现细节

想象一个场景：你花了两周写好了 Agent，代码里到处 import 着 Anthropic 的 SDK 类型。有一天老板说，来，换个 GPT 试试。

你打开项目一看，消息类型用的是 Anthropic 的、事件解析用的是 Anthropic 的、连错误处理都跟 Anthropic 的 API 绑死了。要改动很麻烦。

这就是为什么 LLM 客户端需要做一层封装。上层代码只认你自己定义的类型：消息、流式事件、Token 用量。底下到底调的是 Claude 还是 GPT，上层完全不关心。

6.1 配置只需四个字段

覆盖所有主流供应商，配置上只需要四个字段：

• protocol：走哪家的 API 协议（anthropic / openai）

• model：指定模型

• base_url：端点地址

• api_key：认证

6.2 封装层就是翻译官

封装层干的事情说白了就是翻译。

往外发请求时，把你的统一类型翻译成对应供应商的格式；收到响应时，再翻译回来。

用伪代码表达这个设计：

// 你自己定义的类型（上层代码只用这些）
Message { role, content }
StreamEvent { type, text?, usage?, error? }
Usage { inputTokens, outputTokens }

// 你的客户端（根据 protocol 分发到不同的后端实现）
class LLMClient:
    constructor(protocol, model, baseURL, apiKey)

    function streamChat(systemPrompt, messages) -> Stream<StreamEvent>:
        // 1. 把自定义 Message 转成对应供应商的格式
        // 2. 调用对应的流式 API
        // 3. 把供应商的事件转成自定义 StreamEvent
        // 4. 通过异步流返回给调用方

从调用方视角看，用起来就这么几行：

events = client.streamChat(systemPrompt, messages)
for event in events:
    if event.type == "text":
        print(event.text)          // 逐词打印
    if event.type == "done":
        print(event.usage)         // 显示 token 用量
    if event.type == "error":
        handleError(event.error)   // 处理错误

不管底层走的是 Anthropic 协议还是 OpenAI 协议，调用方的代码完全一样。

用户想换模型，改一下配置文件就行，代码一行不用动。

七、从单轮到多轮：无状态 API 的上下文管理

到目前为止讨论的都是单次 API 调用：发一个请求，拿一个回复，结束。但对于一个 Coding Agent 来说，多轮对话是基本能力——用户描述需求，Agent 问几个澄清问题，然后开始执行，这个过程天然就是多轮的。

那多轮对话是怎么实现的？

答案可能出乎你意料：每次调 API，把完整的对话历史发过去。就这么简单。

Claude API 没有什么会话 ID 让服务器记住之前的对话。每次你发请求，都要把从第一轮到最新一轮的所有消息打包发送。模型靠这些历史消息来理解上下文：

第1轮请求: [user: "写个快排"]
第2轮请求: [user: "写个快排", assistant: "好的...(完整回复)", user: "改成泛型版本"]
第3轮请求: [user: "写个快排", assistant: "好的...", user: "改成泛型版本", assistant: "...", user: "加上单测"]

你需要在客户端维护完整的消息列表，每次用户发消息、模型回复，都要记录下来。Token 消耗随对话轮数线性增长这件事前面已经算过了，后续章节会处理上下文压缩，目前用最简单的全量发送策略。

八、消息模型：为什么需要两层设计

前面定义了面向 API 的消息结构：role + content。但光这两个字段够用吗？

想想 Agent 运行过程中会产生哪些信息：用户输入、模型回复、启动时的欢迎语、API 调用失败的错误信息，后面还会有工具调用记录。这些东西的角色各不相同，光 user 和 assistant 两种根本不够分。

再想想流式接收的场景。模型的回复是一个字一个字蹦出来的，这条 assistant 消息在接收过程中算什么状态？接收完了呢？中间断了呢？一条消息从创建到结束，其实是有生命周期的。

API 层那个简单的 role + content 根本表达不了这些。所以我们需要两层消息模型。

8.1 API 层 vs 内部层

最关键的是多了一个状态字段。一条 assistant 消息刚创建时是 streaming，流式接收完毕变成 complete，出错变成 error。

有了状态，UI 就能根据它决定怎么渲染。格式转换的时候也能把 error 状态的回复过滤掉，别发给 API 让模型困惑。

唯一 ID 也很关键。流式接收时，你需要根据 ID 定位到那条正在接收的 assistant 消息，不断追加文本。没有 ID，你就得靠"最后一条 assistant 消息"这种脆弱的假设来定位，后面场景一复杂就会出问题。

九、对话管理器：并发安全与格式转换

有了消息模型，接下来想一个问题：谁来管这些消息？

你可能觉得，搞一个数组往里面 append 不就行了。

但别忘了流式接收的场景：后台正在往一条 assistant 消息里追加文字，同时另一边正在读这个消息列表来渲染。两处同时操作同一个列表，不加保护就是数据竞争。

所以你需要一个对话管理器，把消息列表包起来，内部保证并发安全（不同语言做法不同，有的用锁，有的靠单线程异步天然避免竞争）。外部调用方只需要：

• 添加消息时拿到一个唯一 ID

• 流式更新时根据 ID 追加内容

• 需要渲染时拿一份消息列表的快照

并发的事情全交给管理器。

9.1 最关键的方法：toAPIFormat()

对话管理器里最关键的方法是 toAPIFormat()：把内部层的消息列表转换成 API 层的格式。这个转换看似只是格式映射，但里面藏着不少坑。

**第一步：过滤。**内部层有些消息不该发给 API。system 角色的消息（比如欢迎语）是内部概念，API 有单独的 system prompt 参数，你再发一条 system 过去会让模型困惑。error 状态的 assistant 消息也得过滤掉——你总不希望模型看到一条报错信息然后尝试接着它说。

**第二步：合并。**虽然 Claude API 能自动合并相邻的同角色消息，但客户端主动合并是更好的做法，减少冗余 token，消息结构也更清晰，方便调试。

**第三步：交替校验。**确保首条消息是 user，并且 user/assistant 交替出现。如果过滤掉 system 消息后第一条变成了 assistant，模型可能理解不了上下文。

这个函数一定要写单元测试。空列表、只有一条消息、连续三条 user、第一条是 system，这些边界情况都得覆盖到。消息格式越规范，模型的理解越准确，后续调试也越轻松。

十、流式更新与多轮协作：把所有东西串起来

前面分别讲了流式响应和多轮对话两个概念。在真实的 Agent 里，它们要配合起来工作。整个流程的伪代码如下：

function sendMessage(userText):
    // 1. 用户消息加入对话管理器
    conversation.addMessage({
        role: "user",
        content: userText,
        status: "complete"
    })

    // 2. 创建一条空的 assistant 消息，状态为 streaming
    assistantId = conversation.addMessage({
        role: "assistant",
        content: "",
        status: "streaming"
    })

    // 3. 把完整对话历史转换成 API 格式
    apiMessages = conversation.toAPIFormat()

    // 4. 异步调用 LLM，流式更新 assistant 消息
    stream = llmClient.streamChat(systemPrompt, apiMessages)
    for event in stream:
        if event.type == "text":
            conversation.updateMessage(assistantId, appendText(event.text))
        if event.type == "done":
            conversation.updateMessage(assistantId, setComplete(event.usage))
        if event.type == "error":
            conversation.updateMessage(assistantId, setError(event.error))

注意第 2 步：先创建一条空的 assistant 消息，拿到 ID，后续通过这个 ID 不断追加内容。这样不管 UI 怎么渲染，数据层的更新逻辑都是一样的。

第 4 步的异步执行方式取决于你用的框架。有的用 Command 模式，有的用 async/await，有的用回调。形式不同，但核心模式一样：后台消费流式事件，实时更新对话管理器里的消息。

整个多轮对话的节奏就是：用户输入 → 加入历史 → 转换格式 → 调 API → 流式更新 → 等待下一轮输入。每一轮都带着完整的对话历史，模型就记住了之前说过的话。

总结

调通 LLM API，套上对话管理支持多轮对话。回顾一下核心要点：

1. Messages 格式：user 和 assistant 交替出现，响应的 content 永远是数组。这两个惯例会深刻影响 Agent 循环里的消息管理逻辑。

2. 流式响应：SSE 事件序列（message_start → content_block_start → content_block_delta → content_block_stop → message_delta → message_stop）是处理流式响应的基本框架。

3. 封装外部依赖：暴露领域语义，隐藏实现细节。定义自己的消息和事件类型，把 SDK 藏在内部。将来换供应商，上层代码完全不受影响。

4. LLM API 是无状态的：多轮对话全靠客户端维护消息历史，每次请求发送完整对话。

5. 消息模型分两层：API 层只有 role + content，内部层增加了状态、ID、元数据。

6. 格式转换：负责过滤、合并、交替校验，把内部状态转成 API 能用的干净格式。这个函数必须写单元测试。