DeepSeek Harness 源码深潜（中）——工具、提示词与插件内核（附DSH源码）

DeepSeek Harness中篇源码解析，拆解工具系统、提示词组装与插件内核，附DSH源码，助你掌握Agent工业级实现。
核心内容：
1. 工具系统：function calling的工业级实现（第4章）
2. 提示词组装：模型调用的“总装车间”（第5章）
3. Cordis插件内核：DSH的立身之本（第6章）

系列：DeepSeek Harness（dsh）源码教程 · 中篇（第 4-7 章）

上篇：从 API 到 Agent 心脏（第 0-3 章）

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

上篇我们拆完了 Agent 的心脏：事件溯源的日志系统、turn/step 状态机、工具调度器。

但心脏只是泵血。中篇进入让 Agent 真正"能干活、可替换、可扩展"的部分：

1. 第 4 章：工具系统——function calling 的工业级实现
2. 第 5 章：提示词组装——模型看到世界的"总装车间"
3. 第 6 章：Cordis 插件内核——dsh 的立身之本
4. 第 7 章：能力缝 seam——可替换的能力接口

如果说上篇是骨架，中篇就是肌肉与神经。

第 4 章 工具系统：FC 的工业级实现

4.1 从手写 dispatch 到工业工具系统

tools = [{”type”: ”function”, ”function”: {”name”: ”get_weather”, ”parameters”: {...}}}]def dispatch(name, args):    if name == ”get_weather”:        return get_weather(**args)

这个写法的问题：

1. 校验靠手写：get_weather(city=123) 这种类型错要自己 if/else
2. 没有把关：任何人都能调任何工具——delete_all() 被误调谁拦？
3. 没有流水线：遥测、限流、权限检查要自己塞进 dispatch
4. 作用域不分：主 agent 和子 agent 共用同一批工具
5. 结果不可追溯：调用结果怎么和日志关联？

dsh 的答案就是本章内容。

4.2 defineTool：声明式定义 + 自动校验

export function defineTool(  options: DefineToolOptions,): ToolDefinition {  const parameters = parameterSchemaSpecToJsonSchema(options.parameters)  const validate = (args: unknown): string[] =>    validateJsonSchemaValue(parameters, args, '')  return {    name: options.name,    description: options.description,    parameters,    async execute(args, exec) {      const violations = validate(args)      if (violations.length > 0) throw new ToolArgsError(violations)      return userExecute(args as InferArgs, exec)    },  }}

关键点：ToolDefinition里execute是被包装过的——你写的 userExecute 永远在参数校验通过之后才被调用。

4.3 真实例子：tool-bash 的三层防御

层次 1：参数 schema

interface BashToolArgs {  command: string  description: string      // 必填——”为什么执行”（可审计）  timeoutMs?: number  workdir?: string  run_in_background?: boolean}

层次 2：业务校验

function validateBashArgs(args: BashToolArgs): void {  if (args.command.trim().length === 0) {    throw new Error('invalid command: expected a non-empty string')  }  // timeoutMs 必须是正有限数  // sandbox_permissions ⇔ justification 必须配对}

为什么 schema 之外还要手写校验？JSON Schema 表达不了"两个字段必须配对出现"这种跨字段约束。

层次 3：动态生成的工具描述

function bashDescription(backgroundEnabled: boolean, escalationModes: readonly SandboxMode[]): string {  // 把当前沙箱模式、后台执行可用性写进描述}

工具描述不是静态字符串，是运行时生成的——环境变了，模型看到的工具说明就变了。

4.4 五段执行流水线

tools/pre-execute   （瀑布事件：允许 / 拒绝 / 询问）      ↓tools/execute       （调度器分发）      ↓tools/post-execute  （结果后处理）      ↓tools/result        （结果观测）

每一个阶段都是一个Cordis 事件，任何插件都能在流水线上插一脚：

1. 安全插件监听 pre-execute：rm -rf / → 拒绝
2. 遥测插件监听 result：统计每次调用的耗时/token
3. 沙箱插件监听 pre-execute：检查参数是否越权

设计精髓：工具自己不管"能不能调"，只管"怎么干活"。安全策略在流水线上。

4.5 TOOL_RUNTIME_SCHEDULER：决策与执行分离

第 3 章的 tool-calls.ts 里出现了一个常量 TOOL_RUNTIME_SCHEDULER：

ctx.provide(TOOL_RUNTIME_SCHEDULER, {  prepare(exec)        // 进入流水线（跑 pre-execute），返回 dispatch / post-result / final-result  dispatch(prepared)   // 真正执行工具函数  finalize(exec, result)  // 结果后处理  finish(exec, result)    // 直接收尾})

const prepared = await ctx.tools[TOOL_RUNTIME_SCHEDULER].prepare(call.exec)switch (prepared.kind) {  case 'dispatch':    promise = ctx.tools[TOOL_RUNTIME_SCHEDULER].dispatch(prepared.exec)  case 'post-result':  case 'final-result':}

为什么要套这一层？prepare() 里跑了 pre-execute 瀑布——监听器可能直接给出结果（"被策略拦下了"），此时根本不需要执行工具。

4.6 执行模式：并行还是串行？

defineTool({  ...,  isConcurrencySafe: true,   // 如 read_file：可并行  // 默认 false：如 bash：必须串行})

为什么"可并行"要工具自己声明？

1. 读文件可以并行，写文件不能——只有工具作者知道语义
2. 猜错了后果严重：两个并行 bash 都改环境变量，结果不可预测
3. 声明是"契约"，调度器按契约执行

4.7 作用域隔离

ctx.tools.register(tool, { scope: agentId })   // 只给这个 agent 注册ctx.tools.register(tool)                        // 全局注册

主 agent 能调"创建子任务"，子 agent 只能调"读写文件"——最小权限在 agent 世界落地。

4.8 Python 对照：带流水线和并发声明的工具系统

class ToolRuntime:    def __init__(self):        self.tools = {}        self.pre_execute_hooks = []    def register(self, tool_def: dict, scope: str = ”*”):        self.tools[(scope, tool_def[”name”])] = tool_def    def get(self, scope: str, name: str):        return self.tools.get((scope, name)) or self.tools.get((”*”, name))    def add_pre_execute_hook(self, hook):        self.pre_execute_hooks.append(hook)    async def prepare(self, scope: str, name: str, args: dict):        tool = self.get(scope, name)        if not tool:            return {”kind”: ”rejected”, ”reason”: ”tool not found”}        for hook in self.pre_execute_hooks:            decision = hook(name, args)            if decision:                return {”kind”: ”rejected”, ”reason”: decision}        return {”kind”: ”dispatch”, ”tool”: tool}    async def dispatch(self, prepared, args: dict):        return prepared[”tool”][”execute”](args)def define_tool(runtime: ToolRuntime, name: str, description: str,                parameters: dict, concurrency_safe: bool = False,                scope: str = ”*”):    def decorator(func):        def execute(args: dict):            for key, spec in parameters.get(”properties”, {}).items():                if spec.get(”required”) and key not in args:                    raise ValueError(f”缺少参数: {key}”)            return func(**args)        runtime.register({            ”name”: name, ”description”: description,            ”parameters”: parameters, ”execute”: execute,            ”concurrency_safe”: concurrency_safe,        }, scope)        return func    return decoratorrt = ToolRuntime()@define_tool(rt, ”read_file”, ”读取文件（可并行）”,             {”type”: ”object”, ”properties”: {”path”: {”type”: ”string”, ”required”: True}}},             concurrency_safe=True)def read_file(path: str): return f”[内容] {path}”@define_tool(rt, ”delete_file”, ”删除文件（危险）”,             {”type”: ”object”, ”properties”: {”path”: {”type”: ”string”, ”required”: True}}})def delete_file(path: str): return f”[已删除] {path}”rt.add_pre_execute_hook(lambda name, args: f”禁止执行 {name}” if name == ”delete_file” else None)import asyncioasync def main():    print(”决策:”, await rt.prepare(”*”, ”delete_file”, {”path”: ”/etc/passwd”}))    prepared = await rt.prepare(”*”, ”read_file”, {”path”: ”a.py”})    print(”执行:”, await rt.dispatch(prepared, {”path”: ”a.py”}))asyncio.run(main())

对照 dsh 的差距：dsh 的瀑布是带 next() 委托语义的 Cordis 事件；校验是完整 JSON Schema 引擎；scope 是分层作用域。但"决策与执行分离 + 瀑布把关 + 并发声明"三个核心已实现。

4.9 本章小结

1. defineTool = 声明式 schema + 自动校验 + 动态描述
2. 五段流水线是事件驱动的，安全策略是插件不是代码
3. TOOL_RUNTIME_SCHEDULER 分离"决策"与"执行"
4. 并发安全是工具的声明，调度器不猜
5. 作用域隔离实现 agent 级最小权限

第 5 章 提示词组装：从字符串拼接到"总装车间"

5.1 你现在的做法，和它的三个致命伤

system_prompt = f”””你是一个智能助手。当前工作目录：{cwd}可用工具：{”, ”.join(tool_names)}规则：{rules}”””

三个问题：

1. 顺序靠手排：新增一段提示词要手工决定插在哪
2. 没有"来源"概念：拼出来的字符串不知道每段来自哪个插件
3. 改动是整体性的：想换某一段等于重拼整个字符串

dsh 的答案：把"提示词"变成注册表 + 总装线。

5.2 section 注册表

export interface PromptSection {  readonly name: string          // 唯一名——重名注册直接抛错  readonly order: number         // 排序权重  readonly text: string | ((context) => string)  readonly complete?: boolean    // ”我就是整个系统提示词”（独占模式）}

规则：每个插件只声明自己的片段，不知道也不关心别人。组装时框架负责排序、拼接、冲突检测。

5.3 组装产物：PromptAssembly——四路输入的总装

export interface PromptAssembly {  sections: AssembledSection[]     // 静态/半静态规则  contexts: AssembledContext[]     // 动态上下文  tools: ToolSchema[]              // 工具 schema  variables: Record   // 模板变量}

组装（assemble）和渲染（render）是两步。组装产生结构化的 PromptAssembly，渲染才把它变成字符串。

5.4 变量插值为什么放最后

PromptSection.text 里可以写 {{variable}}，但插值是渲染阶段的事：

// 组装时：只是把文本解析出来，保留 {{var}} 原样// 渲染时：renderPrompt(assembly) 才把 {{var}} 替换成 assembly.variables 里的值

为什么？sections/contexts/tools 三个阶段都可能贡献变量，如果组装时就插值，顺序耦合就出现了。

5.5 complete 语义

readonly complete?: boolean// 若某 section 标记 complete=true：//   组装仍跑 waterfall//   但最终只保留这一个 section 作为系统提示词// 多个 complete 同时生效 → 组装失败

为什么需要它？有些场景要求"整个系统提示词是我说了算"。complete 是显式的整体替换开关，冲突直接报错，不会静默覆盖。

5.6 Python 对照：带组装/渲染两阶段的提示词系统

from dataclasses import dataclass, fieldimport re@dataclassclass Section:    name: str    order: int    text: str | callable    complete: bool = False@dataclassclass Assembly:    sections: list[dict] = field(default_factory=list)    contexts: list[str] = field(default_factory=list)    tools: list[dict] = field(default_factory=list)    variables: dict = field(default_factory=dict)class SystemPrompt:    def __init__(self):        self._sections: list[Section] = []    def add(self, s: Section):        if any(x.name == s.name for x in self._sections):            raise ValueError(f”重复 section: {s.name}”)        if s.complete and any(x.complete for x in self._sections):            raise ValueError(”多个 complete section 冲突”)        self._sections.append(s)    def assemble(self, context: dict, tools: list[dict]) -> Assembly:        sections = []        for s in sorted(self._sections, key=lambda x: x.order):            text = s.text(context) if callable(s.text) else s.text            sections.append({”name”: s.name, ”text”: text})        completes = [x for x in sections if any(            s.name == x[”name”] and s.complete for s in self._sections)]        if completes:            sections = completes        return Assembly(sections=sections, tools=tools,                        variables=context.get(”variables”, {}))    def render(self, assembly: Assembly) -> str:        parts = [s[”text”] for s in assembly.sections]        if assembly.tools:            parts.append(”可用工具: ” + ”, ”.join(t[”name”] for t in assembly.tools))        text = ”nn”.join(parts)        for key, value in assembly.variables.items():            text = re.sub(r”{{s*” + key + r”s*}}”, str(value), text)        return textsp = SystemPrompt()sp.add(Section(”identity”, -100, ”你是自动化 agent。”))sp.add(Section(”persona”, 0,               lambda ctx: f”你是{ctx['deployment']}的助手，今天是{{{{date}}}}。”))sp.add(Section(”rules”, 150, ”调用工具前必须说明目的。”))asm = sp.assemble({”deployment”: ”工厂质检”, ”variables”: {”date”: ”2026-08-14”}},                  [{”name”: ”read_file”}, {”name”: ”search”}])print(sp.render(asm))

对照 dsh 的差距：dsh 的 text 函数接收 AssembleContext（带 scope/signal），支持 agent 级隔离；工具 schema 是完整 JSON Schema。但"组装/渲染两阶段 + 变量后插值 + complete 独占 + 冲突显性化"四个核心已实现。

5.7 本章小结

1. 系统提示词 = sections + contexts + tools + variables 四路输入的总装
2. 组装与渲染分离：结构化中间产物可被插件检查改写
3. 变量后插值：解耦"谁提供值"与"谁使用值"
4. text 可以是函数：每次请求现场生成
5. complete：显式整体替换，冲突直接报错

第 6 章 一切皆插件：Cordis 微内核

6.1 Cordis 只做三件事

加载（依赖解析、拓扑排序）卸载（副作用逆序回滚）事件（waterfall / serial / emit 三种语义）

"连 agent loop 都是插件"意味着：dsh 里没有"内核"——所有能力都是平级插件。想换主循环？写个插件替换 ctx.agentLoop。想换模型？换个适配器插件。

6.2 插件三要素

export const name = 'tool-bash'export const inject = ['tools', 'shell', 'systemPrompt', 'shellEnv']export function apply(ctx: Context): void {  ctx.tools.register(bashTool)}

inject 不是装饰，是契约：Cordis 加载插件前会解析依赖图，缺依赖的插件根本不加载。

6.3 三类事件语义

// ① waterfall：监听器必须调用 next() 才能放行ctx.emit('tools/pre-execute', data, (decision) => { })// 权力：可以拦截、可以修改// ② serial：按注册顺序执行，但不能改写结果ctx.emit('agent/turn-stopping', { turn, signal })// 权力：可以感知、可以追加副作用// ③ emit：异步通知，监听器互不干扰ctx.emit('session/event', event)// 权力：只能旁观

Cordis 用事件模式把"权力"显式化：要拦截用 waterfall，要感知用 emit。

6.4 作用域（scope）

ctx.on('tools/pre-execute', handler, { scope: agentId })ctx.provide('llm', impl, { scope: agentId })

scope 是"多 agent 世界的防火墙"——每个 agent 有自己独立的插件视角。

6.5 卸载回滚

ctx.provide('llm', impl)          // → 记录: 卸载时删除 'llm'ctx.on('tools/pre-execute', fn)   // → 记录: 卸载时移除监听器ctx.tools.register(tool)          // → 记录: 卸载时注销工具// 卸载时：逆序执行回滚栈

为什么逆序？后注册的往往依赖先注册的。逆序回滚保证依赖关系不被破坏。

6.6 Python 对照：带三类事件和回滚的插件容器

import asyncioclass Cordis:    def __init__(self):        self._services = {}        self._listeners = {}        self._rollbacks = []    def provide(self, name, impl, scope=”*”):        key = (name, scope)        self._services[key] = impl        self._rollbacks.append(lambda: self._services.pop(key, None))    def get(self, name, scope=”*”):        return self._services.get((name, scope)) or self._services.get((name, ”*”))    def on(self, event, handler, scope=”*”):        self._listeners.setdefault((event, scope), []).append(handler)        self._rollbacks.append(            lambda: self._listeners[(event, scope)].remove(handler))    def _collect(self, event, scope):        return (self._listeners.get((event, scope), [])                + self._listeners.get((event, ”*”), []))    async def waterfall(self, event, data, scope=”*”, default=None):        for handler in self._collect(event, scope):            result = await handler(data)            if result is not None:                return result        return default    async def serial(self, event, data, scope=”*”):        for handler in self._collect(event, scope):            await handler(data)    def emit(self, event, data, scope=”*”):        for handler in self._collect(event, scope):            asyncio.create_task(handler(data))    def load(self, plugin):        for dep in plugin.get(”inject”, []):            if self.get(dep) is None:                raise RuntimeError(f”{plugin['name']} 缺少依赖 {dep}”)        plugin[”apply”](self)        self._rollbacks.append(lambda: print(f”[卸载] {plugin['name']}”))    def unload_all(self):        for fn in reversed(self._rollbacks):            fn()

对照真实 Cordis 的差距：真 Cordis 有完整的异步生命周期、依赖图拓扑排序、next() 委托链、作用域的正式分层。但三类事件语义、作用域回退、逆序回滚三个骨架已实现。

6.7 本章小结

1. Cordis = 加载 + 卸载 + 事件
2. 三类事件语义 = 三种权力：waterfall 把关、serial 收尾、emit 旁观
3. scope = 多 agent 世界的防火墙
4. 卸载逆序回滚 = 无孤儿状态

第 7 章 能力缝（seam）——可替换的能力

7.1 什么是 seam：衣服的接缝

一件衣服换袖子，不会把整件衣服重做——因为接缝（seam）把袖子和其他部分解耦了。

Service Definition（接口声明）   ← 接缝本身      ↑ 实现               ↑ 使用Service Provider（实现）         Consumer（消费者）

以 ctx.fs 为例：

Consumer 只依赖接口——这是 seam 的全部秘密。

7.2 import type 强制解耦

import type { } from '@deepseek-ai/dsh-fs'   // ← 只 import 接口（type-only！）export const inject = ['fs', 'tools', 'systemPrompt']export function apply(ctx: Context): void {  const readTool = defineTool({    name: 'read_file',    async execute(args, exec) {      return ctx.fs.read(args.path, exec)   // 调用接口，不知道背后是谁    },  })  ctx.tools.register(readTool)}

import type是关键词——tool-fs 只引入类型，不引入任何 Provider 实现。编译期就保证了 Consumer 与 Provider 解耦。

7.3 换 Provider = 搬家

拆开看：

1. ctx.fs 有多个 Provider（local/sandbox/e2b）
2. ctx.subprocess 有多个 Provider（local/e2b）
3. ctx.shell 通过 ctx.subprocess 执行
4. ctx.lsp 也通过 ctx.subprocess 启动

所以：把subprocess和fs的 Provider 从 local 换成 e2b，shell、terminal、lsp全部自动跟着去远程。

7.4 isolate realm

realm 是比 scope 更严格的服务隔离：scope 是"按 agent 划分视角"，realm 是"一个 agent 完全拥有自己的服务实例"。

主 agent 的 ctx.llm 配置 A 模型，子 agent 的 ctx.llm 配置 B 模型——同名的服务，不同的 realm，各自独立。

7.5 Python 对照：带 seam 的架构升级

from abc import ABC, abstractmethodclass Shell(ABC):    @abstractmethod    def run(self, cmd: str) -> str: ...class BashLocal(Shell):    def run(self, cmd: str) -> str:        import subprocess        return subprocess.run(cmd, shell=True, capture_output=True, text=True).stdoutclass BashSandbox(Shell):    def run(self, cmd: str) -> str:        if ”rm” in cmd:            raise PermissionError(f”[沙箱] 拒绝危险命令: {cmd}”)        return f”[沙箱执行] {cmd} → ok”class BashRemote(Shell):    def run(self, cmd: str) -> str:        return f”[远程执行] {cmd} → ok”class ToolBash:    def __init__(self, shell: Shell):        self._shell = shell    def execute(self, cmd: str) -> str:        return self._shell.run(cmd)CONFIG = {”provider”: ”sandbox”}def make_tool() -> ToolBash:    provider = CONFIG[”provider”]    shell = {”local”: BashLocal, ”sandbox”: BashSandbox,             ”remote”: BashRemote}[provider]()    return ToolBash(shell)CONFIG[”provider”] = ”local”print(make_tool().execute(”echo hi”))CONFIG[”provider”] = ”sandbox”print(make_tool().execute(”echo hi”))try:    make_tool().execute(”rm -rf /”)except PermissionError as e:    print(”被拦截:”, e)CONFIG[”provider”] = ”remote”print(make_tool().execute(”echo hi”))

对照 dsh 的差距：dsh 的 Provider 是插件（通过 cordis 配置加载，可热插拔），realm/scope 提供运行时隔离。但"接口定义 → Provider 注册 → 配置切换 → 业务不变"这条链已跑通。

7.6 本章小结

1. seam = 接口声明 + Provider + Consumer 三角色
2. import type 在编译期强制解耦
3. 换 Provider = 搬家：fs/subprocess 换 Provider，shell/lsp/terminal 全部跟随
4. isolate realm：agent 拥有自己的服务实例

中篇小结

四章下来，你已经掌握了让 Agent 能干活、可替换、可扩展的机制：

1. 工具系统：schema 驱动、五段流水线、决策与执行分离、并发声明、作用域隔离
2. 提示词组装：四路输入、组装/渲染分离、变量后插值、complete 独占
3. Cordis 插件内核：三类事件语义、scope、卸载回滚
4. 能力缝 seam：接口定义 + Provider + Consumer，换 Provider = 搬家

下篇预告：模型适配器、多 Agent subagent、Python 迷你运行时实战。

DSH源码下载链接：

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

登录查看剩余 70% 内容