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

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})

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

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

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

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

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)    }  }}

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

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

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

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

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}

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}

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

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

import asynciofrom dataclasses import dataclass @dataclassclass ToolCall:    name: str    args: dict class 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”: []})()

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