ds4.c 学习笔记

Appearance

Part 5: 代码走读 ​

1. HTTP 服务器 ​

追踪 1：动态缓冲区与 HTTP 响应 ​

buf 家族（动态字符串缓冲区） ​

c
// `buf` 家族（约 109-155 行）: 自增长堆缓冲区

// 扩容（2 倍几何增长）
static void buf_reserve(buf *b, size_t add) {
    while (b->len + add + 1 > b->cap) {
        b->cap = b->cap ? b->cap * 2 : 256;   // 初始 256，之后翻倍
        b->data = realloc(b->data, b->cap);
    }
}

// 追加数据
static void buf_append(buf *b, const void *p, size_t n) {
    buf_reserve(b, n);
    memcpy(b->data + b->len, p, n);
    b->len += n;
    b->data[b->len] = '\0';                   // 始终以 null 结尾
}

// 格式化追加（两次 va_copy: 先测量长度，再写入）
static void buf_printf(buf *b, const char *fmt, ...) {
    va_list ap, ap2;
    va_start(ap, fmt);
    va_copy(ap2, ap);
    int n = vsnprintf(NULL, 0, fmt, ap);       // 测量长度
    buf_reserve(b, n);
    vsnprintf(b->data + b->len, n + 1, fmt, ap2);  // 实际写入
    b->len += n;
    va_end(ap); va_end(ap2);
}

// 所有权转移（调用者负责 free）
static char *buf_take(buf *b) {
    char *data = b->data;
    memset(b, 0, sizeof(*b));                  // 重置为空状态
    return data;                               // 调用者 free()
}

HTTP 响应 ​

c
// `http_response()` 中（约 3080 行）: 发送完整 HTTP/1.1 响应
static bool http_response(int fd, int code, const char *type,
                           const char *body) {
    const char *status = (code == 200) ? "OK"
                       : (code == 400) ? "Bad Request"
                       : (code == 404) ? "Not Found"
                       : (code == 429) ? "Too Many Requests"
                       : "Internal Server Error";
    buf_printf(&b, "HTTP/1.1 %d %s\r\n", code, status);
    buf_printf(&b, "Content-Type: %s\r\n", type);
    buf_printf(&b, "Content-Length: %zu\r\n", strlen(body));
    buf_puts(&b, "Connection: close\r\n\r\n");
    buf_puts(&b, body);
    write_all(fd, b.data, b.len);
}

// `http_error()` 中（约 3097 行）: JSON 错误响应
static bool http_error(int fd, int code, const char *msg) {
    // {"error":{"message":"...","type":"invalid_request_error"}}
    buf_printf(&b, "{\"error\":{\"message\":");
    json_encode_string(&b, msg);
    buf_puts(&b, ",\"type\":\"invalid_request_error\"}}");
    http_response(fd, code, "application/json", b.data);
}

追踪 1b：Context Length 错误处理（协议标准化） ​

参数名自适应 ​

c
// 根据请求类型返回不同的错误参数名
static const char *context_length_error_param(const request *r) {
    if (!r) return "prompt";
    if (r->api == API_RESPONSES) return "input";
    return r->kind == REQ_COMPLETION ? "prompt" : "messages";
}

上下文越界检测 ​

c
// 检查 prompt tokens 是否 >= context size
// ds4_session_sync() 拒绝 prompt->len >= ctx_size，因为生成至少需要一个空闲 slot
static bool request_exceeds_context(const request *r, int ctx_size) {
    return r && r->prompt.len >= ctx_size;
}

协议感知错误响应 ​

c
// 根据请求的 API 类型（OpenAI vs Anthropic）发送不同格式的错误
static bool http_error_context_length_exceeded(int fd, const request *r,
                                               int n_prompt_tokens,
                                               int ctx_size) {
    char msg[160];
    snprintf(msg, sizeof(msg),
             "Prompt has %d tokens, but the configured context size is %d tokens",
             n_prompt_tokens, ctx_size);

    if (r && r->api == API_ANTHROPIC) {
        // Anthropic 格式: {"type":"error","error":{...}}
        buf_puts(&b, "{\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",...");
    } else {
        // OpenAI 格式: {"error":{"message":...,"code":"context_length_exceeded"}}
        buf_puts(&b, "{\"error\":{\"message\":...,\"code\":\"context_length_exceeded\",...");
        // 包含 "param" 字段（messages/input/prompt）
        json_escape(&b, context_length_error_param(r));
    }
    // 两种格式都包含 n_prompt_tokens 和 n_ctx
    http_response(fd, 400, "application/json", b.ptr);
}

client_main 中的集成（提前拦截） ​

c
// 在 client_main() 中，请求解析后、入队前执行检查
static void *client_main(void *arg) {
    // ... 读取 HTTP 请求 ...
    // ... parse_chat_request() / parse_anthropic_request() ...

    // 提前拦截 context length 超限
    if (request_exceeds_context(&req, ctx_size)) {
        http_error_context_length_exceeded(fd, &req, req.prompt.len, ctx_size);
        request_free(&req);
        goto done;       // 直接关闭连接，不浪费 worker 时间
    }

    // ... enqueue(job) → worker 处理推理 ...
}
请求处理管线（新增早期检查）:

HTTP 请求到达
    │
    ▼ 读取请求
    │
    ▼ 解析 JSON（parse_chat_request / parse_anthropic_request）
    │
    ▼ ★ context length 检查（新增）
    │   └── 超限 → 400 错误，连接关闭
    │
    ▼ enqueue(job)
    │
    ▼ worker 取出 → generate_job → 推理 + SSE

追踪 2：SSE 流式推送 ​

Server-Sent Events 协议 ​

c
// `sse_headers()` 中（约 3110 行）: 建立 SSE 连接
static bool sse_headers(int fd) {
    const char *hdr =
        "HTTP/1.1 200 OK\r\n"
        "Content-Type: text/event-stream\r\n"
        "Cache-Control: no-cache\r\n"
        "Connection: close\r\n\r\n";
    return write_all(fd, hdr, strlen(hdr));
}

// `sse_chunk()` 中（约 3119 行）: 发送一个 SSE 数据块
static bool sse_chunk(int fd, const request *r,
                       const char *id, const char *text,
                       const char *finish) {
    buf_puts(&b, "data: ");                    // SSE 格式前缀
    // 根据 API 类型格式化
    if (r->kind == REQ_CHAT) {
        // OpenAI chat completion chunk:
        // {"id":"...","object":"chat.completion.chunk",
        //  "choices":[{"delta":{"content":"text"}}]}
        buf_printf(&b, "{\"id\":\"%s\",\"object\":\"chat.completion.chunk\",...", id);
    } else {
        // Text completion chunk 格式
    }
    buf_puts(&b, "\n\n");                      // SSE 消息结束标记
    write_all(fd, b.data, b.len);
}

// `sse_done()` 中（约 3174 行）: 流结束
static bool sse_done(int fd, const request *r,
                      const char *id, int prompt_tokens,
                      int completion_tokens) {
    if (r->stream_include_usage)
        sse_usage_chunk(fd, r, id, ...);       // 发送 token 使用统计
    write_all(fd, "data: [DONE]\n\n", 14);     // 终止标记
}
SSE 流示例:

→ sse_headers(fd)
← HTTP/1.1 200 OK
← Content-Type: text/event-stream
←
← data: {"id":"chatcmpl-123","choices":[{"delta":{"content":"Hello"}}]}
←
← data: {"id":"chatcmpl-123","choices":[{"delta":{"content":" world"}}]}
←
← data: {"id":"chatcmpl-123","choices":[{"delta":{},"finish_reason":"stop"}]}
←
← data: [DONE]
←

追踪 3：JSON 解析器（手写递归下降） ​

核心解析函数 ​

c
// `json_ws()` 中（约 160 行）: 跳过空白
static void json_ws(const char **p) {
    while (**p == ' ' || **p == '\t' || **p == '\n' || **p == '\r')
        (*p)++;
}

// `json_string()` 中（约 209 行）: 解析 JSON 字符串（处理所有转义序列）
static bool json_string(const char **p, char **out) {
    if (**p != '"') return false;
    (*p)++;                                    // 跳过开头引号
    buf b = {0};
    while (**p != '"') {
        if (**p == '\\') {                     // 转义序列
            (*p)++;
            switch (**p) {
                case '"':  buf_putc(&b, '"');  break;
                case '\\': buf_putc(&b, '\\'); break;
                case '/':  buf_putc(&b, '/');  break;
                case 'n':  buf_putc(&b, '\n'); break;
                case 't':  buf_putc(&b, '\t'); break;
                case 'u':  // \uXXXX Unicode 转义
                    uint32_t cp = parse_hex4(*p);
                    // 处理 UTF-16 代理对
                    if (cp >= 0xd800 && cp <= 0xdbff) {
                        uint32_t cp2 = parse_hex4(*p + 2);  // \uXXXX
                        cp = 0x10000 + ((cp - 0xd800) << 10) + (cp2 - 0xdc00);
                    }
                    utf8_put(&b, cp);
                    break;
            }
        } else {
            buf_putc(&b, **p);                 // 普通字符
        }
        (*p)++;
    }
    (*p)++;                                    // 跳过结尾引号
    *out = buf_take(&b);                       // 返回堆分配字符串
    return true;
}

// `json_number()` 中（约 253 行）: 解析 JSON 数字
static bool json_number(const char **p, double *out) {
    char *end;
    *out = strtod(*p, &end);                   // 标准库转换
    if (end == *p) return false;
    *p = end;
    return true;
}

JSON 解析模式 ​

递归下降解析器结构:

json_object:
  '{' json_ws ( json_string ':' json_value (',' json_string ':' json_value)* )? '}'

json_array:
  '[' json_ws ( json_value (',' json_value)* )? ']'

json_value:
  json_string | json_number | json_object | json_array |
  "true" | "false" | "null"

ds4_server 只实现了项目实际用到的子集（字符串、数字、跳过值），
不追求完整的 JSON 规范兼容

追踪 4：服务器架构 — Job 队列 ​

server 结构体 ​

c
// `server` 结构体定义（约 4717 行）: 服务器状态
struct server {
    ds4_engine  *engine;            // 推理引擎（只读共享）
    ds4_session *session;           // 推理会话（互斥使用）
    int default_tokens;             // 默认生成 token 数
    kv_disk_cache kv;               // KV 磁盘缓存
    tool_memory tool_mem;           // 工具调用内存
    bool disable_exact_dsml_tool_replay;

    pthread_mutex_t tool_mu;        // 工具调用锁
    pthread_mutex_t mu;             // job 队列锁
    pthread_cond_t  cv;             // job 队列条件变量
    pthread_cond_t  clients_cv;     // 客户端等待条件
    job *head;                      // 队列头
    job *tail;                      // 队列尾
    bool stopping;                  // 停止标志
};

生产者-消费者模式 ​

c
// `enqueue()` 中（约 7482 行）: 入队（client_main 调用）
static bool enqueue(server *s, job *j) {
    pthread_mutex_lock(&s->mu);
    if (s->stopping) { pthread_mutex_unlock(&s->mu); return false; }
    j->next = NULL;
    if (s->tail) s->tail->next = j;    // 尾部追加
    else         s->head = j;           // 队列为空
    s->tail = j;
    pthread_cond_signal(&s->cv);        // 通知 worker
    pthread_mutex_unlock(&s->mu);
    return true;
}

// `dequeue()` 中（约 7495 行）: 出队（worker_main 调用）
static job *dequeue(server *s) {
    pthread_mutex_lock(&s->mu);
    while (!s->head && !s->stopping)
        pthread_cond_wait(&s->cv, &s->mu);  // 等待新 job
    if (s->stopping && !s->head) {
        pthread_mutex_unlock(&s->mu);
        return NULL;                         // 服务器关闭
    }
    job *j = s->head;
    s->head = j->next;
    if (!s->head) s->tail = NULL;            // 队列变空
    pthread_mutex_unlock(&s->mu);
    return j;
}

// `worker_main()` 中（约 7510 行）: Worker 主循环
static void *worker_main(void *arg) {
    server *s = arg;
    for (;;) {
        job *j = dequeue(s);
        if (!j) return NULL;                 // 关闭信号
        generate_job(s, j);                  // 执行推理
        j->done = true;
        pthread_cond_signal(&j->cv);         // 通知 client_main
    }
}
架构图:

client_main(fd=5)  client_main(fd=6)  client_main(fd=7)
    │ enqueue          │ enqueue          │ enqueue
    ▼                  ▼                  ▼
  ┌──────────────────────────────────────────────┐
  │          Job Queue (FIFO 链表)                │
  │   head → [job_5] → [job_6] → [job_7] ← tail │
  └──────────────────────────────────────────────┘
                        │ dequeue
                        ▼
                   worker_main
                        │
                   generate_job(s, job)
                        │
                   ds4_session_sync + sample loop
                        │
                   SSE 流式推送 → fd → 网络 → 用户

追踪 5：连接处理与 accept 循环 ​

client_main ​

c
// `client_main()` 中（约 7678 行）: 处理单个客户端连接
static void *client_main(void *arg) {
    client_arg ca = *(client_arg *)arg;
    free(arg);
    server *s = ca.s;
    int fd = ca.fd;

    // 读取 HTTP 请求
    char *request = read_http_request(fd);

    // 路由
    if (GET "/v1/models")
        send_models(fd, s->engine);
    else if (GET "/v1/models/deepseek-v4-flash")
        send_model(fd, s->engine);
    else if (POST "/v1/chat/completions" || POST "/v1/completions")
        handle_chat(fd, s, request);
    else if (POST "/v1/messages")
        handle_anthropic(fd, s, request);
    else
        http_error(fd, 404, "Not found");

    close(fd);
    return NULL;
}

main 函数 ​

c
// `main()` 中（约 8089 行）: 服务器入口
int main(int argc, char **argv) {
    signal(SIGPIPE, SIG_IGN);                   // 忽略 SIGPIPE（断开的 write）
    // 安装优雅退出信号处理
    struct sigaction sa = { .sa_handler = stop_signal_handler };
    sigaction(SIGINT, &sa, NULL);
    sigaction(SIGTERM, &sa, NULL);

    parse_options(argc, argv, &opts);
    ds4_engine_open(&engine, &opts);             // 加载模型
    server s = { .engine = engine, ... };
    ds4_threads_init();                          // 启动线程池

    // 创建 worker 线程（推理线程）
    for (int i = 0; i < n_workers; i++)
        pthread_create(&workers[i], NULL, worker_main, &s);

    // accept 循环
    int listen_fd = listen_on(port);
    while (!s.stopping) {
        int fd = accept(listen_fd, NULL, NULL);
        if (fd < 0) continue;
        // 为每个连接创建独立线程
        client_arg *ca = malloc(sizeof(*ca));
        ca->s = &s; ca->fd = fd;
        pthread_t t;
        pthread_create(&t, NULL, client_main, ca);
        pthread_detach(t);                      // 自动回收
    }
}
完整服务器生命周期:

1. main() 启动
2. 加载模型 (81GB mmap, ~10s)
3. 创建 worker 线程 (等待 job)
4. listen on port 8080
5. accept 循环:
   ├─ 连接到来 → pthread_create(client_main)
   ├─ client_main 读取请求 → enqueue(job)
   ├─ worker dequeue → generate_job
   ├─ generate_job: sync → sample loop → SSE push
   └─ 完成后 close(fd)
6. SIGINT → stop_signal_handler → stopping=true
7. worker 线程退出 → join → 清理

2. KV Cache 持久化 ​

追踪 1：SHA1 哈希实现 ​

SHA-1 压缩函数 ​

c
// 行 5192: 标准 SHA-1 压缩（处理一个 64 字节块）
static void sha1_transform(sha1_ctx *c, const uint8_t block[64]) {
    uint32_t w[80];
    // 前 16 个字直接从输入读取
    for (int i = 0; i < 16; i++)
        w[i] = (block[i*4]<<24) | (block[i*4+1]<<16)
             | (block[i*4+2]<<8) | block[i*4+3];
    // 扩展到 80 个字
    for (int i = 16; i < 80; i++)
        w[i] = rol32(w[i-3] ^ w[i-8] ^ w[i-14] ^ w[i-16], 1);
    //               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  ^^^
    //               四个字的异或                       左旋 1 位

    uint32_t a = c->h[0], b = c->h[1], c_ = c->h[2];
    uint32_t d = c->h[3], e = c->h[4];
    // 注: 局部变量用 c_ 避免与参数 c 冲突

    // 80 轮运算，分 4 段
    for (int i = 0; i < 80; i++) {
        uint32_t f, k;
        if (i < 20)       { f = (b & c_) | (~b & d);  k = 0x5a827999; }
        else if (i < 40)  { f = b ^ c_ ^ d;            k = 0x6ed9eba1; }
        else if (i < 60)  { f = (b & c_) | (b & d) | (c_ & d); k = 0x8f1bbcdc; }
        else               { f = b ^ c_ ^ d;            k = 0xca62c1d6; }

        uint32_t temp = rol32(a, 5) + f + e + k + w[i];
        e = d; d = c_; c_ = rol32(b, 30); b = a; a = temp;
    }
    c->h[0] += a; c->h[1] += b; c->h[2] += c_;
    c->h[3] += d; c->h[4] += e;
}
SHA-1 处理流程:

输入消息 (任意长度)
    │
    ▼ 填充: 追加 0x80 + 零字节 + 64 位大端长度
    │
    ▼ 分割为 64 字节块
    │
    ├── sha1_transform(block_0) → 更新 h[0..4]
    ├── sha1_transform(block_1) → 更新 h[0..4]
    └── ...
    │
    ▼ 输出: h[0]‖h[1]‖h[2]‖h[3]‖h[4] = 20 字节哈希值

初始化与最终化 ​

c
// 行 5233: 标准初始值
static void sha1_init(sha1_ctx *c) {
    c->h[0] = 0x67452301;
    c->h[1] = 0xefcdab89;
    c->h[2] = 0x98badcfe;
    c->h[3] = 0x10325476;
    c->h[4] = 0xc3d2e1f0;
    c->len = 0;
}

// 行 5260: 填充 + 最终化
static void sha1_final(sha1_ctx *c, uint8_t out[20]) {
    // 追加 0x80 填充
    // 对齐到 56 mod 64
    // 追加 64 位大端位长度
    // 输出 20 字节哈希值
}

追踪 2：缓存条目序列化格式 ​

二进制头部布局 ​

c
// 行 5539: 填充固定头部
static void kv_fill_header(uint8_t h[KV_CACHE_FIXED_HEADER],
                            uint8_t quant_bits, uint8_t reason,
                            uint8_t ext_flags,
                            uint32_t tokens, uint32_t hits,
                            uint32_t ctx_size,
                            uint64_t created_at, uint64_t last_used,
                            uint64_t payload_bytes) {
    // 字节 0-2: magic ('D','S','4')
    // 字节 3:   version
    // 字节 4:   quant_bits (2 或 4)
    // 字节 5:   reason (store 触发原因)
    // 字节 6:   ext_flags
    // 字节 7-10:  tokens (LE uint32)
    // 字节 11-14: hits (LE uint32)
    // 字节 15-18: ctx_size (LE uint32)
    // 字节 19-26: created_at (LE uint64)
    // 字节 27-34: last_used (LE uint64)
    // 字节 35-42: payload_bytes (LE uint64)
}

头部二进制布局 ​

偏移  大小  字段            说明
0     3     magic           "DS4" 魔数
3     1     version         版本号
4     1     quant_bits      量化位数 (2 或 4)
5     1     reason          存储触发原因
6     1     ext_flags       扩展标志
7     4     tokens          token 数量 (LE)
11    4     hits            缓存命中次数 (LE)
15    4     ctx_size        上下文长度 (LE)
19    8     created_at      创建时间戳 (LE)
27    8     last_used       最后使用时间 (LE)
35    8     payload_bytes   负载大小 (LE)
---固定头部结束---

43    var   text_bytes      原始文本
?     var   session_data    序列化的 session 状态

读取头部 ​

c
// 行 5560: 从文件读取并验证头部
static bool kv_read_header(FILE *fp, kv_entry *e, uint32_t *text_bytes) {
    uint8_t h[KV_CACHE_FIXED_HEADER];
    if (fread(h, 1, KV_CACHE_FIXED_HEADER, fp) != KV_CACHE_FIXED_HEADER)
        return false;
    if (h[0] != 'D' || h[1] != 'S' || h[2] != '4') return false;
    // 使用 le_get32 / le_get64 读取小端字段
    e->quant_bits = h[4];
    e->tokens     = le_get32(h + 7);
    e->hits       = le_get32(h + 11);
    // ...
}

追踪 3：磁盘缓存打开 / 关闭 / 刷新 ​

缓存初始化 ​

c
// 行 5723: 打开/创建磁盘缓存
static bool kv_cache_open(kv_disk_cache *kc, const char *dir,
                           uint64_t budget_mb,
                           bool reject_different_quant,
                           kv_cache_options opt) {
    mkdir(dir, 0755);                           // 创建缓存目录
    kc->dir = strdup(dir);
    kc->budget = budget_mb * 1024 * 1024;       // MB → 字节
    kc->enabled = true;
    kc->reject_different_quant = reject_different_quant;
    kv_cache_evict(kc, NULL, NULL);              // 先清理超预算条目（无保护 SHA）
}

缓存刷新（扫描磁盘目录） ​

c
// 行 5603: 从磁盘重建内存索引
static void kv_cache_refresh(kv_disk_cache *kc) {
    // 清空内存中的条目列表
    for (int i = 0; i < kc->n_entries; i++)
        free(kc->entries[i].text);
    kc->n_entries = 0;

    // 扫描目录
    DIR *d = opendir(kc->dir);
    struct dirent *de;
    while ((de = readdir(d))) {
        // 只处理 40 字符的 SHA-1 hex 文件名
        if (strlen(de->d_name) != 40) continue;
        // 读取文件头部
        kv_entry e;
        kv_read_entry_file(kc, de->d_name, &e);
        // 推入内存数组
        kc->entries[kc->n_entries++] = e;
    }
    closedir(d);
}
磁盘缓存结构:

kv_cache/
  ├── 2ef7bde608ce5404e97d5f042f95f89f1c232871  ← SHA-1(token_ids) 作为文件名
  │   └── [DS4 头部] + [文本] + [session 序列化数据]
  ├── a3f2c8d...
  └── ...

内存索引:
  entries[n_entries]
    ├── {sha1="2ef7b...", text="Hello world", tokens=42, hits=5, ...}
    ├── {sha1="a3f2c...", text="What is AI",  tokens=38, hits=2, ...}
    └── ...

查找时: SHA-1(新 prompt) 匹配磁盘文件名 → 命中

追踪 4：原子写入与存储 ​

原子写入模式 ​

c
// 行 5921: 将当前 session 的前缀缓存到磁盘
static bool kv_cache_store_live_prefix(server *s,
                                        const ds4_tokens *tokens,
                                        int store_len,
                                        const char *reason) {
    // 前置检查
    if (!kc->enabled) return false;
    if (store_len < min_tokens) return false;
    if (quant_bits != 2 && quant_bits != 4) return false;

    // 关键验证: session checkpoint 必须精确匹配要存储的前缀
    // 这确保不会为未推理的内容创建缓存
    if (s->session->checkpoint.len != store_len) return false;
    for (int i = 0; i < store_len; i++)
        if (s->session->checkpoint.data[i] != tokens->data[i]) return false;

    // 原子写入: 先写临时文件，再 rename
    char tmp[512];
    snprintf(tmp, sizeof(tmp), "%s/%s.tmp.%d", dir, sha1_hex, getpid());
    FILE *fp = fopen(tmp, "wb");

    // 写头部
    kv_fill_header(header, quant_bits, reason, ext_flags, ...);
    fwrite(header, 1, KV_CACHE_FIXED_HEADER, fp);

    // 写文本
    fwrite(text, 1, text_len, fp);

    // 写 session 序列化数据
    ds4_session_save_payload(s->session, fp, ...);

    fclose(fp);

    // 原子重命名（同一文件系统的 rename 是原子的）
    rename(tmp, final_path);
}
原子写入保证:

1. 创建 /kv_cache/abc123.tmp.12345    ← 临时文件（带 PID）
2. 写入全部数据到临时文件
3. fclose()                            ← 确保数据落盘
4. rename(tmp, abc123)                 ← 原子替换

如果步骤 1-3 中间崩溃:
  → 临时文件不完整，但 abc123 不存在或仍是旧版本
  → 不会损坏已有缓存

如果步骤 4 后崩溃:
  → abc123 已是完整的新缓存，没问题

追踪 5：前缀匹配与缓存加载 ​

文本前缀匹配 ​

c
// 行 6068: 查找最长文本前缀匹配
static int kv_cache_find_text_prefix(kv_disk_cache *kc,
                                      const char *prompt_text,
                                      int quant_bits, int ctx_size) {
    kv_cache_refresh(kc);                       // 刷新磁盘索引

    int best_len = 0;
    const char *best_path = NULL;

    for (int i = 0; i < kc->n_entries; i++) {
        kv_entry *e = &kc->entries[i];
        // 过滤条件
        if (e->tokens < kc->min_tokens) continue;
        if (e->ctx_size != ctx_size) continue;
        if (kc->reject_different_quant && e->quant_bits != quant_bits) continue;

        // 检查 prompt_text 是否以 e->text 为前缀
        int tlen = strlen(e->text);
        if (tlen > strlen(prompt_text)) continue;
        if (strncmp(prompt_text, e->text, tlen) == 0) {
            if (tlen > best_len) {
                best_len = tlen;                // 更长匹配
                best_path = e->path;
            }
        }
    }
    return best_len;                            // 返回匹配长度（0 = 未命中）
}

缓存加载入口 ​

c
// 行 6092: 尝试从磁盘缓存加载
static int kv_cache_try_load_text(server *s, const char *prompt_text,
                                   ds4_tokens *effective_prompt,
                                   char **loaded_path_out) {
    // 1. 查找最长前缀匹配
    int match_len = kv_cache_find_text_prefix(&s->kv, prompt_text, ...);
    if (match_len == 0) return 0;               // 未命中

    // 2. 从磁盘文件恢复 session
    FILE *fp = fopen(matched_path, "rb");
    kv_read_header(fp, &entry, &text_bytes);
    ds4_session_load_payload(s->session, fp, ...);
    fclose(fp);

    // 3. 计算 effective_prompt = 匹配前缀之后的剩余部分
    // 如果 prompt_text = "Hello world, how are you?"
    // 匹配的缓存文本 = "Hello world" (11 字符)
    // 则 effective_prompt = ", how are you?"

    return match_len;
}
缓存查找流程:

新请求: "Hello world, how are you?"
    │
    ▼ kv_cache_find_text_prefix
    │
    ├── 扫描所有缓存条目
    ├── "Hello" → 匹配 5 字符
    ├── "Hello world" → 匹配 11 字符 ← 最长匹配
    └── "Goodbye" → 不匹配
    │
    ▼ 加载 "Hello world" 对应的 session
    │
    ▼ effective_prompt = ", how are you?"
    │
    ▼ ds4_session_sync(effective_prompt)
       只需增量推理 ", how are you?" 部分
       跳过了 "Hello world" 的完整 prefill

节省: 11 个 token 的 prefill → 直接恢复 KV cache

3. API 兼容层 ​

追踪 1：OpenAI 请求解析 ​

Chat Completion 请求 ​

c
// 行 1984: 解析 OpenAI /v1/chat/completions 请求
static bool parse_chat_request(ds4_engine *e, server *s,
                                const char *body, int def_tokens,
                                int ctx_size, request *r,
                                char *err, size_t errlen) {
    r->kind = REQ_CHAT;                        // 标记请求类型
    r->api = API_OPENAI;

    // 解析 JSON body 中的字段
    // "model": "deepseek-v4-flash"
    // "messages": [{role, content}, ...]
    // "temperature": 0.7
    // "top_p": 0.9
    // "max_tokens": 1024
    // "stream": true
    // "tools": [...]
    // "thinking": {type: "enabled", budget_tokens: ...}

    if (parse_messages(&p, &r->msgs) == false)  return false;
    if (parse_tools_value(&p, &r->tools_raw, &r->tool_orders) == false)
        return false;

    r->reasoning_effort = DS4_THINK_HIGH;      // 默认高推理
    r->n_predict = def_tokens;
}

Messages 解析 ​

c
// 行 1244: 解析消息数组
static bool parse_messages(const char **p, chat_msgs *msgs) {
    // 期望格式: [{"role":"user","content":"Hello"}, ...]
    json_ws(p); expect('[', p);
    while (**p != ']') {
        json_ws(p); expect('{', p);
        while (**p != '}') {
            char *key;
            json_string(p, &key);
            if (strcmp(key, "role") == 0) {
                json_string(p, &role);
            } else if (strcmp(key, "content") == 0) {
                json_string(p, &content);
            }
            free(key);
            if (**p == ',') (*p)++;
        }
        expect('}', p);
        // 追加到 msgs
        msgs->role[msgs->n] = role;
        msgs->content[msgs->n] = content;
        msgs->n++;
        if (**p == ',') (*p)++;
    }
    expect(']', p);
}

Tools 解析 ​

c
// 行 1211: 解析 OpenAI tools 数组
static bool parse_tools_value(const char **p, char **out,
                               tool_schema_orders *orders) {
    // 每个工具提取 function schema:
    // {"type":"function","function":{"name":"get_weather",
    //  "parameters":{"type":"object",...}}}
    // → 序列化为 JSON 行格式，便于后续 DSML 解析
}

追踪 2：Anthropic 请求解析 ​

Messages API ​

c
// 行 2154: 解析 Anthropic /v1/messages 请求
static bool parse_anthropic_request(ds4_engine *e, server *s,
                                     const char *body, int def_tokens,
                                     int ctx_size, request *r,
                                     char *err, size_t errlen) {
    r->kind = REQ_CHAT;
    r->api = API_ANTHROPIC;                    // 标记 Anthropic API
    // Anthropic 特有字段:
    // "system": "You are helpful"       ← 系统提示在顶层，不在 messages 中
    // "max_tokens": 1024               ← 必填字段
    // "thinking": {type: "enabled", budget_tokens: ...}

    // 解析 messages（Anthropic 格式）
    parse_anthropic_messages(&p, &r->msgs);
}

Anthropic Messages 解析 ​

c
// 行 1494: 解析 Anthropic 消息格式
static bool parse_anthropic_messages(const char **p, chat_msgs *msgs) {
    // Anthropic 消息格式与 OpenAI 不同:
    // content 可以是字符串或数组:
    //   "content": "Hello"                                      ← 简单字符串
    //   "content": [{"type":"text","text":"Hello"},{"type":"image"...}]  ← 内容块数组

    for each message object:
        parse "role" and "content"
        if content is array:
            for each block:
                if type == "text": extract text
                if type == "tool_use": extract tool info
                if type == "tool_result": extract result
}
OpenAI vs Anthropic 格式对比:

OpenAI:
  {"messages":[{"role":"user","content":"Hello"}],
   "tools":[{"type":"function","function":{...}}]}

Anthropic:
  {"messages":[{"role":"user","content":"Hello"}],
   "system":"You are helpful",
   "tools":[{"name":"get_weather","input_schema":{...}}]}

关键差异:
  - system 提示: OpenAI 在 messages 中，Anthropic 在顶层
  - tools 格式: OpenAI 嵌套 function，Anthropic 扁平 name + input_schema
  - content 块: Anthropic 支持 image、tool_use、tool_result 类型

追踪 3：DSML 工具调用解析状态机 ​

参考实现（测试用） ​

c
// 行 3556: 完整 DSML 解析（慢速参考版本）
static dsml_decode_state dsml_decode_state_for_text(
        const char *raw, size_t raw_len) {
    // 搜索 DSML 工具起始标记
    const char *start = dsml_find_tool_start(raw, raw_len);
    if (!start) return DSML_DECODE_TEXT;

    // 遍历 invoke_start / invoke_end 标记
    // 判断当前文本是在 DSML 结构中还是载荷内容中
    // 返回: DSML_DECODE_TEXT / DSML_DECODE_STRUCTURAL / DSML_DECODE_PAYLOAD
}

流式状态机 ​

c
// 行 3646: 增量流式 DSML 追踪器
static void dsml_decode_tracker_update(
        dsml_decode_tracker *dt,
        const char *raw, size_t raw_len) {

    // 三种状态:
    // DSML_TRACK_SEARCH:       扫描工具调用起始标记
    // DSML_TRACK_STRUCTURAL:   在标记边界中
    // DSML_TRACK_DONE:         工具调用块已完成

    // 关键: 保留 hold-back 缓冲区
    // 防止工具标记被 chunk 边界截断
    size_t holdback = dsml_max_tool_start_len();
    // 如果最后 holdback 字节可能是不完整标记，暂不发送
}
流式 DSML 追踪时序:

Chunk 1: "The answer is <dsml:invoke"    ← 可能是不完整标记
  → hold back "<dsml:invoke"             ← 不发送，等下一 chunk

Chunk 2: '_start tool="weather">'         ← 与 holdback 拼接
  → 完整标记: "<dsml:invoke_start tool="weather">"
  → 状态切换到 STRUCTURAL
  → 发送已确认的文本

Chunk 3: '{"city":"SF"}<dsml:invoke_end>'
  → 检测到 invoke_end
  → 状态切换到 DONE
  → 完整的工具调用记录

追踪 4：SSE 流式响应格式化 ​

OpenAI 流式格式 ​

c
// 行 4042: OpenAI 格式 SSE 更新
static bool openai_sse_stream_update(
        int fd, server *s, const request *r,
        const char *id, openai_stream *st,
        const char *raw, size_t raw_len, bool final) {

    // 思考模式检测
    if (st->state == OPENAI_STREAM_THINKING) {
        // 等待思考前缀确认后才发送
        // 前缀字符是特殊的 Unicode 码点
        if (raw[0] == think_marker) {
            st->state = OPENAI_STREAM_THINKING_CONFIRMED;
        } else {
            st->state = OPENAI_STREAM_CONTENT;
        }
    }

    // 格式化 SSE 数据行
    buf_printf(&b, "data: {\"id\":\"%s\",", id);
    buf_printf(&b, "\"object\":\"chat.completion.chunk\",");
    buf_printf(&b, "\"choices\":[{\"index\":0,");
    buf_printf(&b, "\"delta\":{\"content\":");
    json_encode_string(&b, text);
    buf_printf(&b, "},\"finish_reason\":%s}]}\n\n",
               final ? "\"stop\"" : "null");
    write_all(fd, b.data, b.len);
}

Anthropic 流式格式 ​

Anthropic SSE 事件序列:

event: message_start
data: {"type":"message_start","message":{"id":"msg_123",...}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking"}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"thinking_delta","thinking":"Let me..."}}

event: content_block_stop
data: {"type":"content_block_stop","index":0}

event: content_block_start
data: {"type":"content_block_start","index":1,"content_block":{"type":"text"}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"Hello"}}

event: content_block_stop
data: {"type":"content_block_stop","index":1}

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}

event: message_stop
data: {"type":"message_stop"}

追踪 5：生成主循环 ​

generate_job 完整流程 ​

c
// 行 6937: 请求处理主函数
static void generate_job(server *s, job *j) {
    const request *r = &j->r;

    // 1. 计算公共前缀（cache reuse）
    int common = ds4_session_common_prefix(s->session, &prompt);

    // 2. 多级缓存查找
    //    Level 1: 内存 token 精确匹配
    //    Level 2: 内存文本前缀匹配
    //    Level 3: 磁盘文本前缀匹配

    if (common > 0 && common == s->session->checkpoint.len) {
        // 前缀完全匹配，只需增量推理
        ds4_session_sync(s->session, &effective_prompt, ...);
    } else {
        // 需要完整 prefill
        ds4_session_sync(s->session, &prompt, ...);
    }

    // 3. 首次推理获取 logits
    ds4_session_eval(s->session, ...);

    // 4. 采样循环
    for (int i = 0; i < n_predict; i++) {
        // 采样下一个 token
        int token;
        if (r->temperature <= 0)
            token = ds4_session_argmax(s->session);
        else
            token = ds4_session_sample(s->session,
                        r->temperature, r->top_k, r->top_p, r->min_p, &rng);

        // EOS 检查
        if (token == ds4_token_eos(s->engine)) break;

        // 解码 token 文本
        const char *text = ds4_token_text(s->engine, token, &len);

        // DSML 工具调用追踪
        dsml_decode_tracker_update(&dt, text, len);
        // 如果检测到完整工具调用 → 执行并注入结果

        // 思考检查点
        // 如果模型在思考模式，记录 thinking token 计数

        // SSE 流式推送
        if (r->stream) {
            if (r->api == API_OPENAI)
                openai_sse_stream_update(fd, s, r, id, &st, text, len, false);
            else
                anthropic_sse_stream_update(fd, s, r, id, &st, text, len, false);
        }

        // 推理下一个 token
        ds4_session_eval(s->session, token);
    }

    // 5. 流结束
    if (r->stream)
        sse_done(fd, r, id, prompt_tokens, completion_tokens);

    // 6. 可选: 存储到 KV 缓存
    kv_cache_store_current(s, "generate");
}
generate_job 完整时序:

客户端 POST /v1/chat/completions
    │
    ▼ 解析请求 → request 结构体
    │
    ▼ 查找缓存
    │   ├── 内存 token 精确匹配 → 跳过 prefill
    │   ├── 内存文本前缀匹配 → 部分跳过
    │   ├── 磁盘文本前缀匹配 → 从文件恢复
    │   └── 未命中 → 完整 prefill
    │
    ▼ 推理循环:
    │   for each token:
    │     ├── ds4_session_eval → logits
    │     ├── ds4_session_sample → token_id
    │     ├── ds4_token_text → "Hello"
    │     ├── dsml_decode_tracker_update → 工具调用检测
    │     ├── openai_sse_stream_update → SSE 推送
    │     └── ds4_session_eval(token_id) → 下一步
    │
    ▼ sse_done → data: [DONE]
    │
    ▼ kv_cache_store_current → 异步写入磁盘
    │
    ▼ close(fd)

4. ds4_kvstore 模块化 KV 存储 ​

头部/负载格式 ​

KV 持久化逻辑从 ds4_server.c 提取为独立模块 ds4_kvstore.c/h，实现与服务器逻辑解耦：

c
// ds4_kvstore.h — 公共接口
typedef struct ds4_kvstore ds4_kvstore;

// 创建/销毁
ds4_kvstore *ds4_kvstore_open(const char *dir, uint64_t budget_mb, ...);
void ds4_kvstore_close(ds4_kvstore *kvs);

// 存储操作
bool ds4_kvstore_save(ds4_kvstore *kvs, const char *sha1_hex,
                       const uint8_t *header, size_t header_len,
                       const char *text, size_t text_len,
                       ds4_session *session, ...);

// 查找与加载
int ds4_kvstore_find_prefix(ds4_kvstore *kvs, const char *text, ...);
bool ds4_kvstore_load(ds4_kvstore *kvs, const char *sha1_hex,
                       ds4_session *session, ...);

// 驱逐
void ds4_kvstore_evict(ds4_kvstore *kvs, const char *protect_sha1);

// Chat anchor 定位
int ds4_kvstore_chat_anchor_pos(const ds4_tokens *tokens, int len);

头部格式（48 字节固定）：magic KVC + version + model_id + reason + token_count + hits + creation_time + last_used。尾部钩子允许调用方注入协议特定数据（KTM 工具记忆、responses_visible 标记等）。

5. ds4_agent 代理架构 ​

双线程设计 ​

c
// ds4_agent.c — 简化结构
typedef struct {
    pthread_mutex_t mu;
    pthread_cond_t  worker_cv;
    pthread_cond_t  ui_cv;

    // 共享状态
    buf user_input;          // UI → Worker: 用户输入
    buf worker_output;       // Worker → UI: 模型输出
    buf tool_confirmation;   // Worker → UI: 确认请求

    bool worker_busy;
    bool confirmation_result;
    bool stopping;
} agent_shared;

// Worker 线程主循环
static void *agent_worker(void *arg) {
    for (;;) {
        // 等待用户输入
        pthread_mutex_lock(&shared->mu);
        while (!has_input && !shared->stopping)
            pthread_cond_wait(&shared->worker_cv, &shared->mu);

        // 推理 + DSML 流式解析
        // 检测到工具调用 → 请求确认 → 执行工具 → 继续推理
    }
}

// UI 线程主循环
static void agent_ui_loop(agent_shared *shared) {
    for (;;) {
        char *line = linenoise("> ");   // 读取用户输入
        // 传递给 worker
        // 同时监听 worker 输出和确认请求
    }
}

ds4_agent.c 约 9607 行，是项目中仅次于 ds4.c 和 ds4_metal.m 的第三大源文件。