DeepSeek Harness 源码深潜（上）——从 API 到 Agent 心脏（附源码）

调大模型API易，让AI自主干活难？DeepSeek Harness（dsh）开源教程拆解工业级harness，从API到Agent心脏亲手重建。核心内容：1. 从“调API”到Agent运行时的核心机制2. 事件溯源与可追溯性设计3. 轮次状态机的实现原理

系列：DeepSeek Harness（dsh）源码教程 · 上篇（第 0-3 章）

中篇：工具系统、提示词总装与插件内核（第 4-7 章）

下篇：模型适配、多 Agent 与 Python 实战（第 8-10 章）

你有没有过这种感觉：调大模型 API 很简单，几行代码就出结果；但一说到"让 AI 自己干活"——读文件、改代码、跑测试、报错再改——就不知道怎么搭了？

不是不会调 API，是缺一层东西：harness（马具）。

DeepSeek 开源的 dsh 把这层东西做到了工业级。本系列教程的目标：不让你停留在"听说过"，而是把核心机制拆开，再用 Python 亲手重建一遍。

上篇覆盖第 0-3 章：从"调 API"到 agent 运行时、事件溯源、轮次状态机。这是 Agent 的心脏。

一句话总纲：

1. Model（模型）：只会思考。给它文字，回你文字。
2. Harness（马具）：模型之外的一切——工具、循环、记忆、权限、日志。
3. Agent（智能体）：模型套上马具之后，"能干活的东西"。

让 AI 干活的不是模型，是 harness。

第 0 章 总览：dsh 是什么，不是什么

0.1 dsh 不是"更好的模型"，是"模型的操作系统"

很多人第一次看 dsh，会以为它是 DeepSeek 模型的封装。错了。

Model（DeepSeek/V3/R1/任何模型）      ↓   Harness（dsh：20 万行）      ↓   Agent（能读文件、能调工具、能持续迭代）

dsh 不 care 你接哪个模型，它 care 的是：怎么让模型在一个可控、可审计、可扩展的环境里持续运转。

0.2 两条暗线

读 dsh 源码时，抓住两条暗线，所有设计都会变得清晰：

暗线 1：可追溯性。"模型看到的必被记录"——这是 dsh 的运行时断言。日志、把关、中止，全部服务于它。

暗线 2：可替换性。"没有特权核心"——连主循环、模型、UI 都是可换的插件。

0.3 源码地图

packages/├── core/│   ├── session/          ← 会话状态（第 2 章）│   ├── agent-loop/       ← 主循环（第 3 章）│   ├── tools/            ← 工具系统（中篇第 4 章）│   ├── system-prompt/    ← 提示词组装（中篇第 5 章）│   └── agent/            ← agent 公共接口├── cordis/               ← 插件内核（中篇第 6 章）├── fs/  shell/  sandbox/ ← 能力缝（中篇第 7 章）├── llm/llm/              ← 模型适配器（下篇第 8 章）└── subagent/  workflow/  ← 多 agent（下篇第 9 章）

第 1 章 从调 API 到 agent 运行时

1.1 起点：一段"调 API"的代码

from openai import OpenAIclient = OpenAI(api_key=”sk-...”)def chat(system: str, user: str) -> str:    resp = client.chat.completions.create(        model=”deepseek-chat”,        messages=[{”role”: ”system”, ”content”: system},                  {”role”: ”user”, ”content”: user}],    )    return resp.choices[0].message.content

这段代码只做一件事：把消息发给模型，拿回文字。模型是"会思考的马"，代码只是"喂草"的槽子。

1.2 让 AI 自己干活，你缺什么？

"读这个文件 → 找到 bug → 改掉 → 跑测试 → 根据报错再改 → 直到通过"

数一数缺什么：

这 250 行加起来，就是一个最小 harness。

1.3 function calling 循环：harness 雏形

def run_with_tools(task, tools):    messages = [{”role”: ”user”, ”content”: task}]    while True:        resp = client.chat.completions.create(            model=”deepseek-chat”,            messages=messages,            tools=tools,        )        msg = resp.choices[0].message        if not msg.tool_calls:            return msg.content        messages.append({”role”: ”assistant”, ”content”: msg.content})        for call in msg.tool_calls:            result = dispatch(call.function.name, call.function.arguments)            messages.append({”role”: ”tool”,                             ”tool_call_id”: call.id,                             ”content”: result})

这个 20 行的 run_with_tools 已经是最小 harness 骨架——有循环、有回填、有停止判定。但它缺 dsh 那 20 万行里的东西：事件日志、工具把关、错误重试、并发调度、作用域隔离。

dsh 的 40+ 个包，就是把run_with_tools里每个"能跑就行"的点，都做成了工业级。

第 2 章 会话日志：agent 的"数据库"与事件溯源

核心机制：日志是唯一事实来源，一切状态都是它的投影（Event Sourcing）。

2.1 聊天记录不应该是"列表"

你会写：

messages = [{”role”: ”user”, ”content”: ”...”}, ...]

但如果这个列表是唯一状态：

1. 进程崩溃后，能恢复"当时进行到哪一步"吗？❌
2. 能回答"模型在某一步到底看到了什么"吗？❌
3. 能把一次会话"分叉"成两个方向吗？❌
4. 能精确重放"当时发生了什么"吗？❌

普通列表是"状态的快照"，不是"事实的记录"。

2.2 事件溯源：日志是真相，状态是投影

dsh 的 session 采用事件溯源：

1. 只追加（append-only）：每件事都作为不可变事件追加到日志末尾
2. 不修改：已有历史事件永远不被改写
3. 状态即投影：任何当前状态都是从日志重放计算出来的

export interface SessionEvent {  type: string        // 'turn/start' | 'step/start' | 'user/message' | ...  data: any  seq: number         // 全局单调递增序号}

seq 是关键：它是全序的。同样的事件流，任何时候重放，得到完全相同的状态。

一次 turn 会写这样一串事件：

turn/start  step/start    user/message    assistant/chunk*    assistant/message    tool/call    tool/result  step/endturn/end

tool/call 和 tool/result 成对且互相引用——结果事件通过 sourceEventSeqs: [callSeq] 指回调用事件。这是日志里的"外键"。

2.3 核心不变量：模型看到的必被记录

这是 dsh 的运行时断言：

抵达模型请求的一切（系统提示词、历史消息、工具 schema、注入的上下文），都必须能从会话日志重建出来。

因为所有高级能力都是日志的派生品：

会话日志（唯一事实来源）  ├─ deriveMessages()     → 模型历史  ├─ Inbox 构造时重放     → 待处理消息队列  ├─ 持久化后端           → 磁盘（jsonl / sqlite）  ├─ UI 渲染              → 聊天界面  └─ fork/resume/replay   → 新会话或回放

2.4 inbox 也是日志的投影

packages/core/agent/src/inbox.ts 里的待处理消息队列，不是内存里的独立状态，而是日志的投影：

export class Inbox {  constructor(session: Session, ...) {    // 构造时：从日志重放所有 inbox 变更事件    for (const event of session.events.slice(...)) {      if (event.type !== 'agent/inbox/spliced') continue      this.apply(event.data)    }  }}

两个队列：next-turn（等待下一轮的任务）和 next-step（等待下一个步骤的插队输入）。

每次入队/出队，都是一个持久化的 splice 事件，且先写日志、后改内存：

const event = this.session.append('agent/inbox/spliced', splice)const removed = inbox.splice(..., ...event.data.inserted)

隐藏超能力：inbox 状态可重建。进程崩溃后重启，从日志重放，队列原样恢复。

2.5 Python 对照：事件溯源 + 投影

import jsonfrom dataclasses import dataclass, field@dataclassclass Event:    type: str    data: dict    seq: int@dataclassclass Session:    events: list[Event] = field(default_factory=list)    _seq: int = 0    def append(self, type_: str, data: dict) -> Event:        self._seq += 1        ev = Event(type=type_, data=data, seq=self._seq)        self.events.append(ev)        return ev    def derive_messages(self) -> list[dict]:        messages = []        for ev in self.events:            if ev.type == ”user/message”:                messages.append({”role”: ”user”, ”content”: ev.data[”content”]})            elif ev.type == ”assistant/message”:                messages.append({”role”: ”assistant”, ”content”: ev.data[”content”]})            elif ev.type == ”tool/result”:                messages.append({”role”: ”tool”, ”content”: ev.data[”content”]})        return messages    def project_inbox(self) -> dict:        inbox = {”next-turn”: [], ”next-step”: []}        for ev in self.events:            if ev.type != ”agent/inbox/spliced”:                continue            splice = ev.data            target = inbox[splice[”target”]]            del target[splice[”start”]:splice[”start”] + splice.get(”removedCount”, 0)]            target[splice[”start”]:splice[”start”]] = splice[”inserted”]        return inbox    def save(self, path: str):        with open(path, ”w”, encoding=”utf-8”) as f:            for ev in self.events:                f.write(json.dumps({”type”: ev.type, ”data”: ev.data}, ensure_ascii=False) + ”n”)    @classmethod    def load(cls, path: str) -> ”Session”:        s = cls()        with open(path, encoding=”utf-8”) as f:            for line in f:                raw = json.loads(line)                s.append(raw[”type”], raw[”data”])        return s

对照 dsh：你缺的是 seq 全序校验、surface 机制、compaction、多后端。但"日志为真相、状态为投影、恢复即重放"这三个原则，已经实现。

2.6 本章小结

1. session 是事件溯源系统：日志是唯一事实来源
2. 核心不变量"模型看到的必被记录"是运行时断言
3. inbox 队列也是日志的投影——重启恢复，排队消息不丢
4. 日志增长靠 compaction：追加摘要事件，日志依然完整
5. 恢复 = 重放事实，不是还原快照

第 3 章 轮次与步骤：agent 的心脏

核心机制：phase 状态机、inbox 领取语义、工具调用调度器、从日志派生请求。

3.1 turn 和 step

1. step（步骤）：一次模型请求 + 它引发的工具调用
2. turn（轮次）：一个任务的完整过程，含零个或多个 step

一个 turn 的生命周期：

turn/start  step/start    user/message    assistant/chunk*    assistant/message    tool/call*    tool/result*  step/endturn/end

规则：step 内若无工具调用（或无更多工具结果待处理）→ turn 完成。

3.2 phase 状态机

type Phase =  | { kind: 'idle'; lastTurn: number }  | { kind: 'maintenance'; abort: AbortController; lastTurn: number; wakeRequested: boolean }  | { kind: 'running'; abort: AbortController; turn: number; step: number; wakeRequested: boolean }

为什么需要显式状态机？因为 agent 有并发问题：

1. 运行中用户又发消息？→ wakeRequested 锁存，当前结束后再唤醒
2. 运行中插件要执行维护任务？→ runMaintenance：非 idle 直接拒绝
3. 取消后，正在进行的请求怎么停？→ AbortController 信号传播

关键设计：agent 同时只有一个 driver 在跑。新消息要么直接唤醒（idle 时），要么锁存（busy 时）等当前活动结束再处理。

3.3 inbox 的领取（claim）语义

private async preStep(target, position): Promise {  const claimed = this.inbox.claim(target, position.turn)  const assembly = await this.loopCtx.systemPrompt.assemble(...)  const decision = await this.dispatch.waterfall(    'agent/pre-step', { messages: claimed, ... },    () => Promise.resolve({ kind: 'enter', messages: claimed }),  )  return decision}

claim(target, turn): UserMessage[] {  const claimed = this.mutate('next-step', 0, this.nextStep.length, [], false)  if (target === 'next-turn') {    claimed.push(...this.mutate('next-turn', 0, 1, [], false))  }  return claimed}

step 边界会消费掉next-step队列里的所有插队消息，而 turn 边界才会消费next-turn队列里的一条任务。这就是 followup / steer 的落点差异。

3.4 工具调用调度器

工具调用不是"挨个执行"，而是按并发模式分组调度，结果严格按模型顺序提交。

while (next < planned.length) {  const first = planned[next]!  const mode = ctx.tools.executionMode(first.exec).kind  const group = mode === 'parallel' ? planned.slice(next) : [first]  const outcome = await runGroup(ctx, turn, step, group, mode, signal, acceptContext)  next += outcome.consumed}

每个工具声明自己是否并发安全（isConcurrencySafe）。并发安全的工具（如 read_file）可以并行；不安全的（如 bash）必须串行。

runGroup 内部的槽位模型：

const slots = group.map(() => undefined)const commitReady = async () => {  while (committed < group.length) {    const slot = slots[committed]    if (slot === undefined) break    // 提交结果到日志    committed++  }}

结果严格按模型顺序提交：即使调用 2 先完成，也要等调用 1 完成后才能提交。

中止时：

if (aborted) {  for (const call of group.slice(started))    appendSkippedToolCall(session, turn, step, call.block)}

中止 ≠ 抹掉事实，中止 = 记录"未开始"这个事实。回放时，模型看到的依然是完整的"声明了 3 个调用 → 得到 3 个结果"。

3.5 Python 对照：带状态机和工具调度的迷你 loop

import asynciofrom dataclasses import dataclass@dataclassclass ToolCall:    name: str    args: dictclass MiniAgentLoop:    def __init__(self):        self.phase = ”idle”        self.wake_requested = False        self._cancel_event = asyncio.Event()    def send(self, task: str):        if self.phase != ”idle”:            self.wake_requested = True            return        self.phase = ”running”        asyncio.create_task(self._drive(task))    def cancel(self):        self._cancel_event.set()    async def _drive(self, task: str):        try:            await self.turn(task)        finally:            self.phase = ”idle”            if self.wake_requested:                self.wake_requested = False                self.send(”(锁存的任务)”)    async def turn(self, task: str):        step = 0        while True:            self._check_cancel()            step += 1            done = await self.step(task, step)            if done:                return    async def step(self, task: str, step: int) -> bool:        response = await self._llm(task, step)        if not response.tool_calls:            return True        results = await self._schedule(response.tool_calls)        self._pending = results        return False    async def _schedule(self, calls: list[ToolCall]):        results = [None] * len(calls)        for i, call in enumerate(calls):            self._check_cancel()            await asyncio.sleep(0.05)            results[i] = f”[{call.name} 结果]”        return results    def _check_cancel(self):        if self._cancel_event.is_set():            raise asyncio.CancelledError(”agent 被取消”)    async def _llm(self, task, step):        if step == 1:            return type(”R”, (), {”tool_calls”: [ToolCall(”search”, {”q”: ”x”}), ToolCall(”read”, {”p”: ”y”})]})()        return type(”R”, (), {”tool_calls”: []})()

对照 dsh：缺的是 AbortSignal 真正传播、executionMode 按工具判定、槽位 + 顺序提交、中止也写合成结果。但"状态机 + 锁存 + 顺序提交 + 取消传播"四个骨架已搭出。

3.6 本章小结

1. phase 状态机：idle/maintenance/running，保证单 driver 并发安全
2. claim 语义：step 边界消费全部 next-step，turn 边界消费一条 next-turn
3. 请求从日志投影：agent 不持有状态，历史每次现场派生
4. 工具调度器：按工具声明的并发模式分组；结果严格按模型顺序提交
5. AbortSignal 传播链：取消是协作式的，贯穿 agent→step→工具
6. max-tokens 粘性：轮次结果反映最坏情况

上篇小结

三章下来，你已经掌握了 Agent 的心脏：

1. 会话日志（事件溯源）：日志是真相，状态是投影，恢复就是重放
2. 轮次与步骤（状态机）：phase 管理生命周期，工具调度器按并发模式分组、按模型顺序提交

回头再看第 0 章那句话——Model + Harness = Agent。现在你知道了 harness 不是一堆散装功能，而是一套环环相扣的机制：日志支撑循环，循环调度工具。

中篇预告：工具系统、提示词总装车间、Cordis 插件内核、能力缝 seam。

源码下载连接：

https://pan.baidu.com/s/16C5D8_sBubxGWUlFHcRyEg?pwd=fvpf 

登录查看剩余 70% 内容