Agent 实战（一）：打通流式对话，把 SSE 一寸寸拆开看

本节结束时，你会有一个能跟 DeepSeek 流式对话的命令行程序。 不带工具——工具是 M2 的事。这一节只干一件事：把”模型一个字一个字吐字”这件事看透。

0. 本节目标与验收标准

你要建的文件

D:\codeSpace\agent\
├── pom.xml
└── src\main\java\dev\learn\agent\
    ├── Main.java
    └── provider\
        ├── Provider.java
        ├── StreamEvent.java
        ├── Message.java
        ├── ChatRequest.java
        ├── StopReason.java
        ├── Usage.java
        └── deepseek\
            └── DeepSeekProvider.java

验收标准

跑起来之后：

$ mvn -q compile exec:java
模型=deepseek-flash  thinking=false  raw=false
输入问题，空行退出。

你> 用一句话解释什么是递归

递归就是函数自己调用自己，直到碰到一个不用再调用的出口。
[done] reason=END_TURN  in=12 out=28 cache=0

✅ 文字边生成边出现（不是等整段生成完才一次性显示） ✅ 能连续追问，且模型记得上文 ✅ 加 --raw 能看到原始 SSE ✅ 加 --think 能看到灰色的思考过程

1. 先看目标：一次真实请求的原始数据流

在写任何代码之前，先看清楚你要处理的是什么。下面是我用 curl 打出来的真实流（thinking 关闭）：

data: {"id":"6bc6...","object":"chat.completion.chunk","created":1791709705,"model":"deepseek-flash","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}

data: {"id":"6bc6...","choices":[{"index":0,"delta":{"content":"Hi"},"finish_reason":null}]}

data: {"id":"6bc6...","choices":[{"index":0,"delta":{"content":" there"},"finish_reason":null}]}

data: {"id":"6bc6...","choices":[{"index":0,"delta":{"content":" you"},"finish_reason":null}]}

data: {"id":"6bc6...","choices":[{"index":0,"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":3,"total_tokens":13,"prompt_cache_hit_tokens":0,"prompt_cache_miss_tokens":10}}

data: [DONE]

（我把每条里重复的 system_fingerprint、logprobs 等字段删了，方便看重点。）

读一遍，然后回答几个问题：

这一节的全部工作，就是把左边（原始 JSON 帧）翻译成右边（你的程序能用的干净事件）。

2. 概念：SSE 是什么（类比）

普通 HTTP 请求 vs SSE

普通 HTTP 请求像打电话：你说一句，对方把整段话想好、说完才挂断。你听到的永远是完整的一句话。

SSE（Server-Sent Events）像听广播：电台一开始播，你就已经在听了。它说一个字你听到一个字，不用等它说完。

技术上，SSE 就是一个永远不结束的 HTTP 响应体。普通响应是”下完整个文件”，SSE 响应是”文件还在不停地长”，你边读边用。

SSE 的格式（简单到离谱）

data: 一行内容
                    ← 空行 = 一个事件结束
data: 另一行内容
                    ← 空行 = 一个事件结束
data: [DONE]

就三条规则：

1. 每行以 data: 开头（冒号后面有个空格）
2. 空行代表一个事件结束
3. 我们这个场景里，永远只有一行 data，所以一行就是一个事件

💡 因为”一行就是一个事件”，我们的解析器可以极其简单：readLine() 读一行 → 处理一行。不需要状态机，不需要缓冲区。

为什么 Java 里读这个流不会卡住

BufferedReader.readLine() 是阻塞的：没数据来就等着，一来数据就返回。

类比：你守在传送带旁边。传送带上没东西，你就站着等；来一件，你处理一件；再来一件，再处理一件。

所以你不需要开线程、不需要回调地狱、不需要响应式框架。一个朴素的 while 循环就够了：

while ((line = reader.readLine()) != null) {
    处理这一行;
}

这个循环会在服务器关闭连接时结束（readLine() 返回 null）。

3. 第一步：pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
                             http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>dev.learn</groupId>
    <artifactId>agent</artifactId>
    <version>0.1.0</version>
    <packaging>jar</packaging>

    <properties>
        <maven.compiler.release>21</maven.compiler.release>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <jackson.version>2.18.2</jackson.version>
        <junit.version>5.11.4</junit.version>
    </properties>

    <dependencies>
        <!-- 唯一的运行时依赖：JSON 解析 -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>${jackson.version}</version>
        </dependency>

        <!-- 测试。M1 用不上，但先放着，M2 会用到 -->
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- 让我们能直接 mvn exec:java 跑起来 -->
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>exec-maven-plugin</artifactId>
                <version>3.5.0</version>
                <configuration>
                    <mainClass>dev.learn.agent.Main</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

几个说明：

• maven.compiler.release=21 而不是 source/target。这是现代写法，能保证用的是 Java 21 的标准库 API（我们要用 sealed、record、模式匹配的 switch）。
• <packaging>jar</packaging> + exec-maven-plugin：这样 mvn compile exec:java 一条命令就能跑，不用手动配 classpath。
• 没有 Spring、没有 Lombok、没有日志框架。每个依赖都要能说出为什么需要它。

4. 第二步：定义”通用语言”（provider 包）

这三个文件是 Provider 抽象的词汇表。它们不认识 DeepSeek，也不认识任何厂商——它们描述的是”一次模型调用过程中会发生什么”。

4.1 StopReason.java

package dev.learn.agent.provider;

/**
 * 一轮回复"为什么停下来"。
 *
 * <p>这是我们从各家 API 的"方言"里提炼出来的统一说法。
 * DeepSeek/OpenAI 叫 finish_reason，值可能是 "stop" / "tool_calls" / "length"；
 * Anthropic 叫 stop_reason，值可能是 "end_turn" / "tool_use"。
 * 上层代码只认识这个枚举，一辈子不用改。
 */
public enum StopReason {

    /** 模型自然说完了。对方 "stop" */
    END_TURN,

    /** 模型要调工具，等我们把结果喂回去。对方 "tool_calls" */
    TOOL_USE,

    /** 被 max_tokens 截断了，话没说完。对方 "length" */
    MAX_TOKENS,

    /** 其它情况（内容过滤等） */
    OTHER;

    /**
     * 把线上的字符串翻译成我们的枚举。
     * 这个"翻译"就是 Provider 抽象的日常工作——把方言归一成普通话。
     */
    public static StopReason fromWire(String finishReason) {
        if (finishReason == null) {
            return OTHER;
        }
        return switch (finishReason) {
            case "stop" -> END_TURN;
            // "function_call" 是 OpenAI 的旧名字，一起认了
            case "tool_calls", "function_call" -> TOOL_USE;
            case "length" -> MAX_TOKENS;
            default -> OTHER;
        };
    }
}

4.2 Usage.java

package dev.learn.agent.provider;

/**
 * 一次调用的 token 用量。
 *
 * <p>M1 只是显示出来看看。M4 的上下文管理要靠它判断"还剩多少额度、该不该压缩了"。
 *
 * @param promptTokens     输入的 token 数（你发过去的历史 + 系统提示 + 工具定义）
 * @param completionTokens 输出的 token 数（模型生成的，含思考过程的 token）
 * @param cacheHitTokens   命中上下文缓存的输入 token 数（⚠️ 命中部分的单价便宜 50 倍）
 */
public record Usage(int promptTokens, int completionTokens, int cacheHitTokens) {

    /** 拿不到用量时的占位值，避免到处判 null */
    public static final Usage EMPTY = new Usage(0, 0, 0);
}

为什么 promptTokens 值得关注？ 因为它每一轮都在涨。第 1 轮可能 10 个 token，第 20 轮可能 20000 个。这就是 M4 要解决的问题的起点。M1 就把这个数字显示出来，让你亲眼看到它涨。

4.3 StreamEvent.java ★本节最重要的文件

package dev.learn.agent.provider;

/**
 * Provider 和上层之间的"通用语言"。
 *
 * <p>一次模型调用过程中会发生的事情，全部在这里列出来了。上层只认这些事件，
 * 不认任何厂商的原始 JSON。
 *
 * <h3>为什么用 sealed？</h3>
 * {@code sealed} 的意思是"实现这个接口的类型，只能是本文件里列出的这几个，
 * 不许别人再发明新的"。这带来一个实实在在的好处：<b>编译器能帮你在 switch 里查漏</b>。
 * 少写一个分支，编译就不过。后面 Main 里的 switch 一个 default 都不用写，
 * 而且以后给这个接口加一个新事件时，编译器会把所有需要改的地方一个不漏地指出来。
 */
public sealed interface StreamEvent {

    /** 正文增量：模型吐出一小段正文。 */
    record TextDelta(String text) implements StreamEvent {}

    /**
     * 思考增量：thinking 开启时，模型先"想"再答，想的内容从这里出来。
     * M1 用 --think 才能看到；M2 起它还有一个更重要的用途（见铁律 6）。
     */
    record ReasoningDelta(String text) implements StreamEvent {}

    /**
     * 工具调用开始：模型决定调用某个工具，这一片带着 id 和工具名。
     * ⚠️ 每个工具只会发一次这个事件（对应分片里唯一带 id/name 的那一片）。
     */
    record ToolUseStart(String id, String name) implements StreamEvent {}

    /**
     * 工具参数增量：工具的入参 JSON 正在一片片到达。
     * ⚠️ 这里是"碎片"，不是完整 JSON。要按 id 累积起来再解析。
     * M1 不会触发它，但词表先定义好。
     */
    record ToolUseArgsDelta(String id, String partialJson) implements StreamEvent {}

    /** 一轮结束。带上为什么停，以及本次的 token 用量。 */
    record Done(StopReason reason, Usage usage) implements StreamEvent {}

    /** 出错了：网络问题、HTTP 非 200、解析不了的帧…… */
    record Failure(String message) implements StreamEvent {}
}

sealed 到底解决了什么问题？

没有 sealed 的时候，你要处理”事件”，只能写：

if (event instanceof TextDelta) { ... }
else if (event instanceof Done) { ... }
// 忘了写 Failure 分支？编译器不管，运行时静默忽略，你永远不知道。

有了 sealed：

switch (event) {
    case StreamEvent.TextDelta t -> ...
    case StreamEvent.Done d -> ...
    // 少写一个分支 → 编译失败。编译器成了你的护卫。
}

类比：sealed 像一份”只允许这几种”的清单。普通接口是”谁都可以来实现我”，sealed 是”只有名单上的人能进”。代价是你失去了扩展性，收益是编译器能穷举——在”事件类型”这种天生就该封闭的场景里，这笔买卖非常划算。

注意我没有定义 ToolUseEnd。因为流协议里没有”某个工具调用结束”的显式信号——所有工具调用的参数收集完毕的时刻，就是 Done 到达的时刻。不要发明协议里不存在的东西。

5. 第三步：数据模型

5.1 Message.java

package dev.learn.agent.provider;

/**
 * 对话历史里的一条消息。
 *
 * <p>M1 只需要 role + content 两个字段——因为我们还不调工具。
 * M2 会给它加上 toolCalls / toolCallId / reasoningContent 三个字段（见铁律 1 和 6）。
 * <b>这是刻意设计的演进</b>：先跑通最简形态，再按需要长。
 *
 * @param role    角色："system" / "user" / "assistant" / "tool"
 * @param content 文本内容
 */
public record Message(String role, String content) {

    public static Message system(String content) {
        return new Message("system", content);
    }

    public static Message user(String content) {
        return new Message("user", content);
    }

    public static Message assistant(String content) {
        return new Message("assistant", content);
    }
}

💡 为什么要有静态工厂方法？ Message.system("...") 比 new Message("system", "...") 好两点：一是字符串常量不会被写错（写错 "sytem" 编译器不报错，但 API 会），二是以后改内部表示时不用改所有调用点。

5.2 ChatRequest.java

package dev.learn.agent.provider;

import java.util.List;

/**
 * 一次模型请求。
 *
 * <p>为什么不直接传 List&lt;Message&gt;？因为请求还有别的部分——M2 要加工具定义，
 * M4 要加 max_tokens 和缓存断点。把它们收在一个对象里，
 * {@code Provider} 接口的签名就再也不用改了。
 *
 * @param model    模型 ID，例如 "deepseek-flash"
 * @param messages 完整对话历史（含 system）
 * @param thinking 是否开启思考模式。M1 默认关掉——见第 10 节为什么要这样
 */
public record ChatRequest(String model, List<Message> messages, boolean thinking) {

    /** 最常用的一种：不开思考。 */
    public static ChatRequest plain(String model, List<Message> messages) {
        return new ChatRequest(model, messages, false);
    }
}

6. 第四步：Provider 接口

package dev.learn.agent.provider;

import java.util.function.Consumer;

/**
 * 模型提供方的统一接口。
 *
 * <p>这是整个抽象的"插头标准"。左边（Agent / Main）只认 StreamEvent，
 * 右边（DeepSeek / OpenAI / 别的）各自去翻译自己的方言。
 *
 * <p>实现类必须保证：<b>至少发出一个 Done 或一个 Failure，然后返回</b>。
 */
public interface Provider {

    /**
     * 发起一次请求，边收边把事件回调出去。
     *
     * <p><b>这是个阻塞方法</b>：直到流结束（收到 [DONE] 或连接断开）才返回。
     * 调用方不需要处理线程问题——这正是流式 HTTP 在 Java 里的舒适区。
     *
     * @param request 请求内容
     * @param onEvent 每解析出一个事件就调用一次。⚠️ 在本方法返回前，它会在同一个线程上被反复调用。
     */
    void stream(ChatRequest request, Consumer<StreamEvent> onEvent);
}

为什么是 Consumer<StreamEvent> 而不是返回一个 List<StreamEvent>？

因为返回 List 意味着必须等全部收完才能返回——那就把流式变成了非流式，你会在 Hi there you 全部生成完后才看到它。用回调，事件边产生边交付。

为什么不用 Flow / Reactor 那套响应式？ 因为那会把复杂度抬高一个量级，而收益在这个场景是零。Consumer 就够了。能用简单方案的时候不要上框架。

7. 第五步：DeepSeekProvider ★本节的主角

这个类干三件事：拼请求 → 读流 → 翻译事件。

package dev.learn.agent.provider.deepseek;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import dev.learn.agent.provider.ChatRequest;
import dev.learn.agent.provider.Message;
import dev.learn.agent.provider.Provider;
import dev.learn.agent.provider.StopReason;
import dev.learn.agent.provider.StreamEvent;
import dev.learn.agent.provider.Usage;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * DeepSeek 的 Provider 实现。把 DeepSeek 的 SSE 方言翻译成 StreamEvent 普通话。
 */
public class DeepSeekProvider implements Provider {

    private static final String ENDPOINT   = "https://api.deepseek.com/chat/completions";
    private static final String DATA_PREFIX = "data:";
    private static final String DONE_SENTINEL = "[DONE]";

    private final String apiKey;

    /** 打开后把每一行原始 SSE 打到 stdout。学习期强烈建议开。 */
    private final boolean dumpRaw;

    private final HttpClient http;

    /** Jackson 的入口对象。线程安全，可以复用一个。 */
    private final ObjectMapper json = new ObjectMapper();

    public DeepSeekProvider(String apiKey, boolean dumpRaw) {
        this.apiKey = apiKey;
        this.dumpRaw = dumpRaw;
        this.http = HttpClient.newBuilder()
                .connectTimeout(Duration.ofSeconds(30))
                .build();
    }

    // ═══════════════════════════════════════════════════════════════
    //  第一部分：发起请求
    // ═══════════════════════════════════════════════════════════════

    @Override
    public void stream(ChatRequest request, Consumer<StreamEvent> onEvent) {
        try {
            // ⚠️ 用 ofInputStream() 而不是 ofString()。
            //    ofString() 会把整个响应体收完才返回 —— 流式就废了。
            //    ofInputStream() 立刻返回一个"还能继续读"的流。
            HttpResponse<InputStream> response = http.send(
                    buildHttpRequest(request),
                    HttpResponse.BodyHandlers.ofInputStream());

            if (response.statusCode() != 200) {
                String body = new String(response.body().readAllBytes(), StandardCharsets.UTF_8);
                onEvent.accept(new StreamEvent.Failure(
                        "HTTP " + response.statusCode() + " — " + body));
                return;
            }

            readStream(response.body(), onEvent);

        } catch (InterruptedException e) {
            // ⚠️ 被人打断时要恢复中断标志，否则上层感知不到"该收工了"
            Thread.currentThread().interrupt();
            onEvent.accept(new StreamEvent.Failure("请求被中断"));
        } catch (Exception e) {
            onEvent.accept(new StreamEvent.Failure(
                    e.getClass().getSimpleName() + ": " + e.getMessage()));
        }
    }

    private HttpRequest buildHttpRequest(ChatRequest request) {
        return HttpRequest.newBuilder()
                .uri(URI.create(ENDPOINT))
                .timeout(Duration.ofMinutes(10))
                .header("Content-Type", "application/json")
                .header("Accept", "text/event-stream")
                .header("Authorization", "Bearer " + apiKey)
                .POST(HttpRequest.BodyPublishers.ofString(
                        buildBody(request), StandardCharsets.UTF_8))
                .build();
    }

    /**
     * 拼请求体。M1 只有 model / messages / stream / thinking 四样东西。
     * M2 会在这里加上 "tools" —— 那时候你就理解为什么要有 ChatRequest 这个类了。
     */
    private String buildBody(ChatRequest request) {
        ObjectNode body = json.createObjectNode();
        body.put("model", request.model());
        body.put("stream", true);

        // 思考模式开关。DeepSeek 默认是开的，M1 我们显式关掉（见第 10 节）
        body.putObject("thinking")
            .put("type", request.thinking() ? "enabled" : "disabled");

        ArrayNode messages = body.putArray("messages");
        for (Message m : request.messages()) {
            messages.addObject()
                    .put("role", m.role())
                    .put("content", m.content());
        }
        return body.toString();
    }

    // ═══════════════════════════════════════════════════════════════
    //  第二部分：一行一行读流
    // ═══════════════════════════════════════════════════════════════

    private void readStream(InputStream body, Consumer<StreamEvent> onEvent) throws Exception {
        // index → (id, name) 的映射表。
        // 见铁律 3：只有第一片带 id/name，后面的参数碎片只有 index。
        // M1 用不到（没工具），但提前放在这儿，M2 不用改动这个方法的结构。
        Map<Integer, ToolCallIdentity> toolIdentities = new HashMap<>();

        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(body, StandardCharsets.UTF_8))) {

            String line;
            // ⚠️ readLine() 是阻塞的：没数据就等着，来一行就返回一行。
            //    服务器关连接时返回 null，循环自然结束。
            while ((line = reader.readLine()) != null) {

                if (dumpRaw) {
                    System.out.println("[RAW] " + line);
                }

                // SSE 里可能有空行（事件分隔符）和以 ':' 开头的注释/心跳行，跳过
                if (line.isBlank() || line.startsWith(":")) {
                    continue;
                }

                // 我们只关心 data: 开头的行
                if (!line.startsWith(DATA_PREFIX)) {
                    continue;
                }

                String payload = line.substring(DATA_PREFIX.length()).trim();

                // ⚠️ [DONE] 不是 JSON！别拿去 parse。它只是个"没有更多了"的哨兵。
                //    真正的结束信号在上一帧的 finish_reason 上，那时候已经发过 Done 了。
                if (payload.equals(DONE_SENTINEL)) {
                    return;
                }

                handleChunk(payload, toolIdentities, onEvent);
            }
        }
    }

    /** 工具的身份信息（id + 名字）。第一片里拿到的，用来给后面的参数碎片认门。 */
    private record ToolCallIdentity(String id, String name) {}

    // ═══════════════════════════════════════════════════════════════
    //  第三部分：把一帧 JSON 翻译成事件
    // ═══════════════════════════════════════════════════════════════

    private void handleChunk(String payload,
                             Map<Integer, ToolCallIdentity> toolIdentities,
                             Consumer<StreamEvent> onEvent) {

        JsonNode frame;
        try {
            frame = json.readTree(payload);
        } catch (Exception e) {
            onEvent.accept(new StreamEvent.Failure("无法解析的 SSE 帧: " + payload));
            return;
        }

        // 取 choices[0]。⚠️ 用 path() 而不是 get()：
        //    get() 找不到返回 null（你得判空），path() 找不到返回一个"缺失节点"（可以安全地继续调 isXxx()）。
        JsonNode choices = frame.path("choices");
        if (!choices.isArray() || choices.isEmpty()) {
            return;
        }
        JsonNode choice = choices.get(0);

        // ---- ① 增量内容 ----
        JsonNode delta = choice.path("delta");
        if (delta.isObject()) {
            dispatchDelta(delta, toolIdentities, onEvent);
        }

        // ---- ② 结束信号 ----
        // ⚠️ 判据是 finish_reason 变成了"一个真正的字符串"。
        //    流中间它一直是 null（JSON null），只有最后一帧才有值。
        //    用 isTextual() 一个判断就够了：
        //      · 值存在 → TextNode      → isTextual() == true  ✅
        //      · JSON null → NullNode   → isTextual() == false
        //      · 字段不存在 → MissingNode → isTextual() == false
        JsonNode finishReason = choice.path("finish_reason");
        if (finishReason.isTextual()) {
            // 💡 usage 就在同一帧上（铁律 5）。顺手取走，不用等 [DONE]。
            onEvent.accept(new StreamEvent.Done(
                    StopReason.fromWire(finishReason.asText()),
                    readUsage(frame.path("usage"))));
        }
    }

    /** 处理 delta 里的三种可能内容：思考、正文、工具调用。 */
    private void dispatchDelta(JsonNode delta,
                               Map<Integer, ToolCallIdentity> toolIdentities,
                               Consumer<StreamEvent> onEvent) {

        // ⚠️ 用 isTextual() 做守卫，不要直接信 asText() 的返回。
        //    Jackson 里 NullNode.asText() 会返回字面量字符串 "null"（不是空串！），
        //    直接拿去用会让你 println 出一个 "null" 来，而且很难查。
        //    isTextual() 对 NullNode / MissingNode 都是 false，最稳。

        // 思考内容（thinking 开启时才有）
        JsonNode reasoning = delta.path("reasoning_content");
        if (reasoning.isTextual() && !reasoning.asText().isEmpty()) {
            onEvent.accept(new StreamEvent.ReasoningDelta(reasoning.asText()));
        }

        // 正文
        JsonNode content = delta.path("content");
        if (content.isTextual() && !content.asText().isEmpty()) {
            onEvent.accept(new StreamEvent.TextDelta(content.asText()));
        }

        // 工具调用（M1 不会出现；M2 才用得上，但翻译逻辑已经完整了）
        JsonNode toolCalls = delta.path("tool_calls");
        if (toolCalls.isArray()) {
            for (JsonNode toolCall : toolCalls) {
                translateToolCall(toolCall, toolIdentities, onEvent);
            }
        }
    }

    /**
     * 翻译一个工具调用分片（铁律 3）。
     *
     * <p>分片有两种形态：
     * <pre>
     *   形态 A（每个 index 第一次出现，带身份）:
     *     {"index":0,"id":"call_00_EQSI...","type":"function",
     *      "function":{"name":"get_weather","arguments":""}}
     *
     *   形态 B（后续的参数碎片，只有 index 和 arguments）:
     *     {"index":0,"function":{"arguments":"{\""}}
     * </pre>
     * 靠 index 把两者对应起来。
     */
    private void translateToolCall(JsonNode toolCall,
                                   Map<Integer, ToolCallIdentity> toolIdentities,
                                   Consumer<StreamEvent> onEvent) {

        int index = toolCall.path("index").asInt(0);

        // --- 形态 A：带 id，说明是这个工具的第一片 ---
        JsonNode id = toolCall.path("id");
        if (id.isTextual()) {
            String name = toolCall.path("function").path("name").asText("");
            toolIdentities.put(index, new ToolCallIdentity(id.asText(), name));
            onEvent.accept(new StreamEvent.ToolUseStart(id.asText(), name));
        }

        // --- 参数碎片（形态 A 和 B 都可能有，只是 A 里是空串） ---
        JsonNode arguments = toolCall.path("function").path("arguments");
        if (arguments.isTextual() && !arguments.asText().isEmpty()) {
            ToolCallIdentity identity = toolIdentities.get(index);
            if (identity == null) {
                // 理论上不会发生。真发生了说明服务端行为变了，别静默吞掉。
                onEvent.accept(new StreamEvent.Failure(
                        "收到 index=" + index + " 的参数碎片，但从未见过它的 id"));
                return;
            }
            onEvent.accept(new StreamEvent.ToolUseArgsDelta(identity.id(), arguments.asText()));
        }
    }

    private Usage readUsage(JsonNode usage) {
        if (!usage.isObject()) {
            return Usage.EMPTY;
        }
        return new Usage(
                usage.path("prompt_tokens").asInt(0),
                usage.path("completion_tokens").asInt(0),
                usage.path("prompt_cache_hit_tokens").asInt(0));
    }
}

三个值得停下来想的点

① 为什么 isTextual() 而不是 asText() != null？

Jackson 的节点类型和你想的不一样：

⚠️ 第二行是经典的坑：NullNode.asText() 返回的是字符串 "null"，不是空串。你会莫名其妙打印出一个 null 来。用 isTextual() 做守卫可以一次性绕开全部情况。

② 为什么工具的表叫 toolIdentities 而不是直接把分片攒起来？

职责分离：Provider 只负责”翻译”，不负责”累积”。

• 翻译 = 把厂商方言转成 StreamEvent（Provider 的活）
• 累积 = 把一堆 ToolUseArgsDelta 拼成完整的工具调用（Agent 的活，M2 要写）

如果 Provider 把参数攒好再给你，你就学不到”累积”这件事了——而这恰恰是 M2 的核心难点之一。让每一层只做自己该做的事，是这类系统能长期演进的关键。

③ readStream 抛异常，但 stream 会 catch —— 为什么？

因为回调式 API 里不能往外抛异常（调用方的 while 循环还在等你回调）。所有错误都必须变成 Failure 事件走同一条路。 这样上层只需要处理一种错误形态。

8. 第六步：Main（REPL）

package dev.learn.agent;

import dev.learn.agent.provider.ChatRequest;
import dev.learn.agent.provider.Message;
import dev.learn.agent.provider.Provider;
import dev.learn.agent.provider.StreamEvent;
import dev.learn.agent.provider.deepseek.DeepSeekProvider;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

public class Main {

    /** ⚠️ 如果 API 报 "Model Not Exist"，改成你控制台里准确的模型 ID */
    private static final String MODEL = "deepseek-flash";

    private static final String SYSTEM_PROMPT = """
            你是一个简洁的编程助手。
            回答尽量短，不要客套话，不要重复问题。
            """;

    public static void main(String[] args) throws Exception {

        // ⚠️ API key 只从环境变量读。绝不硬编码，绝不提交到 git。
        String apiKey = System.getenv("DEEPSEEK_API_KEY");
        if (apiKey == null || apiKey.isBlank()) {
            System.err.println("请先设置环境变量 DEEPSEEK_API_KEY");
            System.exit(1);
        }

        List<String> flags = List.of(args);
        boolean dumpRaw  = flags.contains("--raw");    // 打印原始 SSE
        boolean thinking = flags.contains("--think");  // 开启思考模式

        Provider provider = new DeepSeekProvider(apiKey, dumpRaw);

        // ── 对话历史。⚠️ 它在整个会话里持续增长，这就是 M4 要处理的东西 ──
        List<Message> conversation = new ArrayList<>();
        conversation.add(Message.system(SYSTEM_PROMPT));

        BufferedReader stdin = new BufferedReader(
                new InputStreamReader(System.in, StandardCharsets.UTF_8));

        System.out.printf("模型=%s  thinking=%s  raw=%s%n", MODEL, thinking, dumpRaw);
        System.out.println("输入问题，空行退出。");

        while (true) {
            System.out.print("\n你> ");
            System.out.flush();

            String input = stdin.readLine();
            if (input == null || input.isBlank()) {
                break;
            }

            conversation.add(Message.user(input));

            // 收集这一轮的正文，结束后写回历史
            StringBuilder reply = new StringBuilder();

            provider.stream(
                    // 💡 List.copyOf 做一次不可变快照：Provider 只看到"此刻"的历史，
                    //    之后我们往 conversation 里追加东西，不会影响正在进行的这轮请求。
                    new ChatRequest(MODEL, List.copyOf(conversation), thinking),
                    event -> render(event, reply));

            // 把助手的回答写回历史 —— 没有这一行，模型下一轮就不记得自己说过什么
            conversation.add(Message.assistant(reply.toString()));
        }

        System.out.println("\n再见。");
    }

    /** 把事件渲染到终端，顺便把正文攒起来。 */
    private static void render(StreamEvent event, StringBuilder reply) {
        // 💡 switch 里没有 default —— 因为 StreamEvent 是 sealed 的，
        //    编译器知道全部可能性，少写一个分支就编译不过。这就是 sealed 的回报。
        switch (event) {
            case StreamEvent.TextDelta t -> {
                System.out.print(t.text());
                System.out.flush();          // ⚠️ flush 很重要，否则会攒在缓冲区里不显示
                reply.append(t.text());
            }

            case StreamEvent.ReasoningDelta r -> {
                // 灰色显示思考过程：ANSI 转义 \u001B[2m ... \u001B[0m
                System.out.print("\u001B[2m" + r.text() + "\u001B[0m");
                System.out.flush();
            }

            case StreamEvent.ToolUseStart s ->
                    System.out.printf("%n[tool] 准备调用 %s (id=%s)%n", s.name(), s.id());

            case StreamEvent.ToolUseArgsDelta a ->
                    System.out.print(a.partialJson());

            case StreamEvent.Done d ->
                    System.out.printf("%n%n[done] reason=%s  in=%d out=%d cache=%d%n",
                            d.reason(),
                            d.usage().promptTokens(),
                            d.usage().completionTokens(),
                            d.usage().cacheHitTokens());

            case StreamEvent.Failure f ->
                    System.err.println("\n[失败] " + f.message());
        }
    }
}

⚠️ System.out.flush() 不能省。终端输出是有缓冲的，不 flush 的话你会看到”模型想完了，文字才一次性冒出来”——那就白做流式了。

9. 跑起来

cd /d/codeSpace/agent

# 编译
mvn -q compile

# 跑
mvn -q compile exec:java

先看原始协议（强烈建议第一次这样跑）

mvn -q compile exec:java -Dexec.args="--raw"

然后随便问一句。你会看到 [RAW] 前缀的原始行。对照第 1 节的那个样例逐行看一遍：

1. 第一行："delta":{"role":"assistant","content":""} ← 声明身份
2. 中间若干行："delta":{"content":"..."} ← 文字碎片
3. 倒数第二行："finish_reason":"stop" 和 "usage":{...} 在一起 ← 结束 + 用量
4. 最后一行：data: [DONE] ← 哨兵

看到这四步都出现了，说明你的解析器是对的。

用一个能看出流式的长回答试

短回答（“hi”）你是看不出流式效果的。问个长的：

你> 请用大约 200 字介绍一下 TCP 三次握手，要详细

文字应该是平滑地一个字一个字冒出来的。如果是一大段突然出现，回去检查是不是漏了 flush()，或者是不是用了 BodyHandlers.ofString()。

10. 打开思考模式看看

mvn -q compile exec:java -Dexec.args="--think"

然后问一个需要想的问题：

你> 9.11 和 9.8 哪个大？

你会看到灰色的思考过程先流出来，然后才是正常颜色的答案。原始帧长这样：

delta: {"role":"assistant","content":null,"reasoning_content":""}
delta: {"content":null,"reasoning_content":"We"}
delta: {"content":null,"reasoning_content":" need"}
...
delta: {"content":"9.8","reasoning_content":null}

注意两件事：

1. 只有一个字段有值。推理时 content 是 null，回答时 reasoning_content 是 null。
2. [done] 那行的 out= 数字包含思考消耗的 token，所以会比你想的大很多。

⚠️ 为什么 M1 默认关掉思考？以及 M2 的坑

不是因为它不好，是因为M1 要一次只学一个变量。流式解析本身就是个新东西，再加一层”思考/回答两股流交织”，调试起来会互相干扰。

但是——你要提前记住这条：

一旦请求带上了 tools 参数，所有历史轮次的 reasoning_content 必须原样回传，否则 API 返回 400。

也就是说 M2 一开始调工具，你就必须：

1. 给 Message 加上 reasoningContent 字段
2. 在 ReasoningDelta 到达时把它攒起来（就像 Main.render 攒 reply 一样）
3. 把攒好的 reasoning_content 放进 assistant 消息里回传

漏了第 3 步的典型症状：M1 好好的，M2 加了工具之后第一个请求就 400，错误信息还看不出原因。这就是为什么我在 M1 就把 ReasoningDelta 事件和 Message 的说明都准备好了。

11. 自检清单

进 M2 之前，逐条确认：

• mvn -q compile 无错误、无警告
• 普通模式能连续对话，且模型记得上一轮（说明历史回写对了）
• 长回答是平滑流出的，不是一次性出现
• --raw 能看到第 1 节描述的四步结构
• [done] 里的 in= 随着对话轮次持续增长（这是 M4 的伏笔）
• --think 能看到灰色思考过程，且 out= 明显变大
• 你能不看文档，口述出 StreamEvent 有哪六个事件、分别什么时候触发
• 你能解释为什么 Provider.stream 里要 catch 而不是往外抛

几个可以自己动手改的练习

1. 加一个 --model 参数，让它能切到 deepseek-v4-pro。这是 Provider 抽象还不需要改动就能做到的事。
2. 把 [done] 的 in= 累积起来，在退出时打印整个会话的总消耗。你会对”上下文增长”有实感。
3. 故意把 body.put("stream", true) 改成 false，看看会发生什么。你会收到一个非 SSE 的 JSON 响应体，你的 data: 前缀判断一个也匹配不上，循环直接结束、什么都不打印。这是一个很好的”沉默失败”案例——想想真实项目里怎么避免（提示：检查 Content-Type 响应头）。

12. 踩坑速查

13. 本节你实际学到了什么

回头看，这一节你亲手写了这些东西：

最重要的收获不是这些代码，而是你现在知道”模型吐字”到底是怎么一回事了。 后面所有东西——工具调用、权限、压缩——都是在这个事件流上加东西。

14. 下一节预告（M2）

M2 你要写 Agent 和 Tool，把 M1 定义的 ToolUseStart / ToolUseArgsDelta 真正用起来：

1. Tool 接口：名字、描述、JSON schema、执行方法
2. 参数累积器：把 ToolUseArgsDelta 的碎片按 id 拼成完整 JSON —— 你会在自己的代码里遇到铁律 3 那个坑
3. 工具并发执行：一次请求可能来两个工具（实测确认过），用虚拟线程并行跑
4. 核心循环：while (finish_reason == TOOL_USE) { 执行 → 回灌 }
5. reasoning_content 回传：不加这一步，第一次调工具就 400

先让 M1 跑起来，跑通了再开 M2。