系列:DeepSeek Harness(dsh)源码教程 · 中篇(第 4-7 章)上篇:从 API 到 Agent 心脏(第 0-3 章)下篇:模型适配、多 Agent 与 Python 实
<div class="detail-content-box has-mask large">
<blockquote style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.55);line-height: 1.8;margin-bottom: 24px;"><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>系列:DeepSeek Harness(dsh)源码教程 · 中篇(第 4-7 章)</span><span><br/></span><span>上篇:从 API 到 Agent 心脏(第 0-3 章)</span><span><br/></span><span>下篇:模型适配、多 Agent 与 Python 实战(第 8-10 章)</span></p></blockquote><hr/><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>上篇我们拆完了 Agent 的心脏:事件溯源的日志系统、turn/step 状态机、工具调度器。</span></p><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>但心脏只是泵血。中篇进入让 Agent 真正"能干活、可替换、可扩展"的部分:</span></p><ul style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span>第 4 章:<span style="font-weight: 500;">工具系统</span>——function calling 的工业级实现</span></section></li><li style="margin-bottom: 0px;"><section><span>第 5 章:<span style="font-weight: 500;">提示词组装</span>——模型看到世界的"总装车间"</span></section></li><li style="margin-bottom: 0px;"><section><span>第 6 章:<span style="font-weight: 500;">Cordis 插件内核</span>——dsh 的立身之本</span></section></li><li style="margin-bottom: 0px;"><section><span>第 7 章:<span style="font-weight: 500;">能力缝 seam</span>——可替换的能力接口</span></section></li></ul><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>如果说上篇是骨架,中篇就是肌肉与神经。</span></p><section style="text-align: center;"><img src="https://pic.imgdb.cn/i/034D0AyxwUgaPrHT6N2ob8.jpg" class="rich_pages wxw-img"/></section><hr/><h2 style="font-size: 17px;font-weight: 500;color: #2B77BF;line-height: 1.8;margin-bottom: 12px;"><span>第 4 章 工具系统:FC 的工业级实现</span></h2><blockquote style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.55);line-height: 1.8;margin-bottom: 24px;"><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>核心机制:<span style="font-weight: 500;">schema 驱动的声明式校验、五段执行流水线、TOOL_RUNTIME_SCHEDULER 调度器、执行模式(并行/串行)、作用域隔离</span>。</span></p></blockquote><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.1 从手写 dispatch 到工业工具系统</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">tools = [{”type”: ”function”, ”function”: {”name”: ”get_weather”, ”parameters”: {...}}}] def dispatch(name, args): if name == ”get_weather”: return get_weather(**args)</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>这个写法的问题:</span></p><ol style="list-style-type: decimal;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">校验靠手写</span>:</span><code><span>get_weather(city=123)</span></code><span> 这种类型错要自己 if/else</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">没有把关</span>:任何人都能调任何工具——</span><code><span>delete_all()</span></code><span> 被误调谁拦?</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">没有流水线</span>:遥测、限流、权限检查要自己塞进 dispatch</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">作用域不分</span>:主 agent 和子 agent 共用同一批工具</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">结果不可追溯</span>:调用结果怎么和日志关联?</span></section></li></ol><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>dsh 的答案就是本章内容。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.2 defineTool:声明式定义 + 自动校验</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">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) }, }}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">关键点:ToolDefinition里execute是被包装过的</span>——你写的 userExecute 永远在参数校验通过之后才被调用。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.3 真实例子:tool-bash 的三层防御</span></h3><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">层次 1:参数 schema</span></span></p><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">interface BashToolArgs { command: string description: string // 必填——”为什么执行”(可审计) timeoutMs?: number workdir?: string run_in_background?: boolean}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">层次 2:业务校验</span></span></p><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">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 必须配对}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">为什么 schema 之外还要手写校验?</span>JSON Schema 表达不了"两个字段必须配对出现"这种跨字段约束。</span></p><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">层次 3:动态生成的工具描述</span></span></p><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">function bashDescription(backgroundEnabled: boolean, escalationModes: readonly SandboxMode[]): string { // 把当前沙箱模式、后台执行可用性写进描述}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">工具描述不是静态字符串,是运行时生成的</span>——环境变了,模型看到的工具说明就变了。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.4 五段执行流水线</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">tools/pre-execute (瀑布事件:允许 / 拒绝 / 询问) ↓tools/execute (调度器分发) ↓tools/post-execute (结果后处理) ↓tools/result (结果观测)</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>每一个阶段都是一个<span style="font-weight: 500;">Cordis 事件</span>,任何插件都能在流水线上插一脚:</span></p><ul style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span>安全插件监听 </span><code><span>pre-execute</span></code><span>:</span><code><span>rm -rf /</span></code><span> → 拒绝</span></section></li><li style="margin-bottom: 0px;"><section><span>遥测插件监听 </span><code><span>result</span></code><span>:统计每次调用的耗时/token</span></section></li><li style="margin-bottom: 0px;"><section><span>沙箱插件监听 </span><code><span>pre-execute</span></code><span>:检查参数是否越权</span></section></li></ul><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">设计精髓:工具自己不管"能不能调",只管"怎么干活"。安全策略在流水线上。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.5 TOOL_RUNTIME_SCHEDULER:决策与执行分离</span></h3><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>第 3 章的 tool-calls.ts 里出现了一个常量 TOOL_RUNTIME_SCHEDULER:</span></p><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">ctx.provide(TOOL_RUNTIME_SCHEDULER, { prepare(exec) // 进入流水线(跑 pre-execute),返回 dispatch / post-result / final-result dispatch(prepared) // 真正执行工具函数 finalize(exec, result) // 结果后处理 finish(exec, result) // 直接收尾})</pre></section><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">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':}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">为什么要套这一层?</span>prepare() 里跑了 pre-execute 瀑布——监听器可能直接给出结果("被策略拦下了"),此时根本不需要执行工具。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.6 执行模式:并行还是串行?</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">defineTool({ ..., isConcurrencySafe: true, // 如 read_file:可并行 // 默认 false:如 bash:必须串行})</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">为什么"可并行"要工具自己声明?</span></span></p><ol style="list-style-type: decimal;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span>读文件可以并行,写文件不能——只有工具作者知道语义</span></section></li><li style="margin-bottom: 0px;"><section><span>猜错了后果严重:两个并行 bash 都改环境变量,结果不可预测</span></section></li><li style="margin-bottom: 0px;"><section><span>声明是"契约",调度器按契约执行</span></section></li></ol><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.7 作用域隔离</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">ctx.tools.register(tool, { scope: agentId }) // 只给这个 agent 注册ctx.tools.register(tool) // 全局注册</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>主 agent 能调"创建子任务",子 agent 只能调"读写文件"——<span style="font-weight: 500;">最小权限在 agent 世界落地</span>。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.8 Python 对照:带流水线和并发声明的工具系统</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">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 decorator rt = 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())</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">对照 dsh 的差距:</span>dsh 的瀑布是带 next() 委托语义的 Cordis 事件;校验是完整 JSON Schema 引擎;scope 是分层作用域。<span style="font-weight: 500;">但"决策与执行分离 + 瀑布把关 + 并发声明"三个核心已实现。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>4.9 本章小结</span></h3><ul style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span>defineTool = 声明式 schema + 自动校验 + 动态描述</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">五段流水线</span>是事件驱动的,安全策略是插件不是代码</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">TOOL_RUNTIME_SCHEDULER</span> 分离"决策"与"执行"</span></section></li><li style="margin-bottom: 0px;"><section><span>并发安全是<span style="font-weight: 500;">工具的声明</span>,调度器不猜</span></section></li><li style="margin-bottom: 0px;"><section><span>作用域隔离实现 agent 级最小权限</span></section></li></ul><hr/><h2 style="font-size: 17px;font-weight: 500;color: #2B77BF;line-height: 1.8;margin-bottom: 12px;"><span>第 5 章 提示词组装:从字符串拼接到"总装车间"</span></h2><blockquote style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.55);line-height: 1.8;margin-bottom: 24px;"><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>核心机制:<span style="font-weight: 500;">组装/渲染分离、变量后插值、complete 语义、组装瀑布</span>。</span></p></blockquote><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>5.1 你现在的做法,和它的三个致命伤</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">system_prompt = f”””你是一个智能助手。当前工作目录:{cwd}可用工具:{”, ”.join(tool_names)}规则:{rules}”””</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>三个问题:</span></p><ol style="list-style-type: decimal;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">顺序靠手排</span>:新增一段提示词要手工决定插在哪</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">没有"来源"概念</span>:拼出来的字符串不知道每段来自哪个插件</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">改动是整体性的</span>:想换某一段等于重拼整个字符串</span></section></li></ol><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>dsh 的答案:<span style="font-weight: 500;">把"提示词"变成注册表 + 总装线</span>。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>5.2 section 注册表</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">export interface PromptSection { readonly name: string // 唯一名——重名注册直接抛错 readonly order: number // 排序权重 readonly text: string | ((context) => string) readonly complete?: boolean // ”我就是整个系统提示词”(独占模式)}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">规则:每个插件只声明自己的片段,不知道也不关心别人。</span>组装时框架负责排序、拼接、冲突检测。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>5.3 组装产物:PromptAssembly——四路输入的总装</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">export interface PromptAssembly { sections: AssembledSection[] // 静态/半静态规则 contexts: AssembledContext[] // 动态上下文 tools: ToolSchema[] // 工具 schema variables: Record // 模板变量}</pre></section><table><thead><tr class="firstRow"><th style="text-align: left;"><section><span>路</span></section></th><th style="text-align: left;"><section><span>内容</span></section></th><th style="text-align: left;"><section><span>生命周期</span></section></th></tr></thead><tbody><tr><td style="text-align: left;"><section><span>sections</span></section></td><td style="text-align: left;"><section><span>规则性文本</span></section></td><td style="text-align: left;"><section><span>基本静态,配置时注册</span></section></td></tr><tr><td style="text-align: left;"><section><span>contexts</span></section></td><td style="text-align: left;"><section><span>动态信息</span></section></td><td style="text-align: left;"><section><span><span style="font-weight: 500;">每次请求现算</span></span></section></td></tr><tr><td style="text-align: left;"><section><span>tools</span></section></td><td style="text-align: left;"><section><span>工具 schema</span></section></td><td style="text-align: left;"><section><span>注册时收集,组装时排序</span></section></td></tr><tr><td style="text-align: left;"><section><span>variables</span></section></td><td style="text-align: left;"><code><span>{{date}}</span></code><section><span> 类占位</span></section></td><td style="text-align: left;"><section><span>渲染时才插值</span></section></td></tr></tbody></table><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">组装(assemble)和渲染(render)是两步。</span>组装产生结构化的 PromptAssembly,渲染才把它变成字符串。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>5.4 变量插值为什么放最后</span></h3><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>PromptSection.text 里可以写 {{variable}},但<span style="font-weight: 500;">插值是渲染阶段的事</span>:</span></p><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">// 组装时:只是把文本解析出来,保留 {{var}} 原样// 渲染时:renderPrompt(assembly) 才把 {{var}} 替换成 assembly.variables 里的值</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">为什么?</span>sections/contexts/tools 三个阶段都可能贡献变量,如果组装时就插值,顺序耦合就出现了。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>5.5 complete 语义</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">readonly complete?: boolean// 若某 section 标记 complete=true:// 组装仍跑 waterfall// 但最终只保留这一个 section 作为系统提示词// 多个 complete 同时生效 → 组装失败</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">为什么需要它?</span>有些场景要求"整个系统提示词是我说了算"。complete 是<span style="font-weight: 500;">显式的整体替换开关</span>,冲突直接报错,不会静默覆盖。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>5.6 Python 对照:带组装/渲染两阶段的提示词系统</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">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 = ”\n\n”.join(parts) for key, value in assembly.variables.items(): text = re.sub(r”\{\{\s*” + key + r”\s*\}\}”, str(value), text) return text sp = 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))</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">对照 dsh 的差距:</span>dsh 的 text 函数接收 AssembleContext(带 scope/signal),支持 agent 级隔离;工具 schema 是完整 JSON Schema。<span style="font-weight: 500;">但"组装/渲染两阶段 + 变量后插值 + complete 独占 + 冲突显性化"四个核心已实现。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>5.7 本章小结</span></h3><ul style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span>系统提示词 = <span style="font-weight: 500;">sections + contexts + tools + variables</span> 四路输入的总装</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">组装与渲染分离</span>:结构化中间产物可被插件检查改写</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">变量后插值</span>:解耦"谁提供值"与"谁使用值"</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">text 可以是函数</span>:每次请求现场生成</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">complete</span>:显式整体替换,冲突直接报错</span></section></li></ul><hr/><h2 style="font-size: 17px;font-weight: 500;color: #2B77BF;line-height: 1.8;margin-bottom: 12px;"><span>第 6 章 一切皆插件:Cordis 微内核</span></h2><blockquote style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.55);line-height: 1.8;margin-bottom: 24px;"><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>核心机制:<span style="font-weight: 500;">三类事件语义(waterfall/serial/emit)</span>、<span style="font-weight: 500;">作用域(scope)</span>、<span style="font-weight: 500;">服务生命周期</span>。</span></p></blockquote><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>6.1 Cordis 只做三件事</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">加载(依赖解析、拓扑排序)卸载(副作用逆序回滚)事件(waterfall / serial / emit 三种语义)</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">"连 agent loop 都是插件"</span>意味着:dsh 里没有"内核"——所有能力都是平级插件。想换主循环?写个插件替换 ctx.agentLoop。想换模型?换个适配器插件。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>6.2 插件三要素</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">export const name = 'tool-bash'export const inject = ['tools', 'shell', 'systemPrompt', 'shellEnv'] export function apply(ctx: Context): void { ctx.tools.register(bashTool)}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">inject 不是装饰,是契约</span>:Cordis 加载插件前会解析依赖图,缺依赖的插件<span style="font-weight: 500;">根本不加载</span>。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>6.3 三类事件语义</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">// ① waterfall:监听器必须调用 next() 才能放行ctx.emit('tools/pre-execute', data, (decision) => { })// 权力:可以拦截、可以修改 // ② serial:按注册顺序执行,但不能改写结果ctx.emit('agent/turn-stopping', { turn, signal })// 权力:可以感知、可以追加副作用 // ③ emit:异步通知,监听器互不干扰ctx.emit('session/event', event)// 权力:只能旁观</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">Cordis 用事件模式把"权力"显式化:要拦截用 waterfall,要感知用 emit。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>6.4 作用域(scope)</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">ctx.on('tools/pre-execute', handler, { scope: agentId })ctx.provide('llm', impl, { scope: agentId })</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">scope 是"多 agent 世界的防火墙"</span>——每个 agent 有自己独立的插件视角。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>6.5 卸载回滚</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">ctx.provide('llm', impl) // → 记录: 卸载时删除 'llm'ctx.on('tools/pre-execute', fn) // → 记录: 卸载时移除监听器ctx.tools.register(tool) // → 记录: 卸载时注销工具 // 卸载时:逆序执行回滚栈</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">为什么逆序?</span>后注册的往往依赖先注册的。逆序回滚保证依赖关系不被破坏。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>6.6 Python 对照:带三类事件和回滚的插件容器</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">import asyncio class 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()</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">对照真实 Cordis 的差距:</span>真 Cordis 有完整的异步生命周期、依赖图拓扑排序、next() 委托链、作用域的正式分层。<span style="font-weight: 500;">但三类事件语义、作用域回退、逆序回滚三个骨架已实现。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>6.7 本章小结</span></h3><ul style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span>Cordis = 加载 + 卸载 + 事件</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">三类事件语义 = 三种权力</span>:waterfall 把关、serial 收尾、emit 旁观</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">scope</span> = 多 agent 世界的防火墙</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">卸载逆序回滚</span> = 无孤儿状态</span></section></li></ul><hr/><h2 style="font-size: 17px;font-weight: 500;color: #2B77BF;line-height: 1.8;margin-bottom: 12px;"><span>第 7 章 能力缝(seam)——可替换的能力</span></h2><blockquote style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.55);line-height: 1.8;margin-bottom: 24px;"><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>核心机制:<span style="font-weight: 500;">seam 三角色</span>、</span><code><span><span style="font-weight: 500;">import type</span></span></code><span><span style="font-weight: 500;"> 强制解耦</span>、<span style="font-weight: 500;">"换 Provider = 搬家"</span>、<span style="font-weight: 500;">isolate realm</span>。</span></p></blockquote><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>7.1 什么是 seam:衣服的接缝</span></h3><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>一件衣服换袖子,不会把整件衣服重做——因为**接缝(seam)**把袖子和其他部分解耦了。</span></p><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">Service Definition(接口声明) ← 接缝本身 ↑ 实现 ↑ 使用Service Provider(实现) Consumer(消费者)</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>以 ctx.fs 为例:</span></p><table><thead><tr class="firstRow"><th style="text-align: left;"><section><span>角色</span></section></th><th style="text-align: left;"><section><span>是什么</span></section></th><th style="text-align: left;"><section><span>真实代码</span></section></th></tr></thead><tbody><tr><td style="text-align: left;"><section><span>Definition</span></section></td><td style="text-align: left;"><code><span>ctx.fs</span></code><section><span> 接口</span></section></td><td style="text-align: left;"><code><span>packages/fs/fs/src/index.ts</span></code></td></tr><tr><td style="text-align: left;"><section><span>Provider</span></section></td><td style="text-align: left;"><section><span>具体实现</span></section></td><td style="text-align: left;"><code><span>fs-local</span></code><section><span>、</span><code><span>fs-sandbox</span></code><span>、</span><code><span>fs-e2b</span></code></section></td></tr><tr><td style="text-align: left;"><section><span>Consumer</span></section></td><td style="text-align: left;"><section><span>模型调用的工具</span></section></td><td style="text-align: left;"><code><span>tool-fs</span></code></td></tr></tbody></table><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">Consumer 只依赖接口</span>——这是 seam 的全部秘密。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>7.2 import type 强制解耦</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">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)}</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">import type是关键词</span>——tool-fs 只引入类型,不引入任何 Provider 实现。<span style="font-weight: 500;">编译期就保证了 Consumer 与 Provider 解耦。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>7.3 换 Provider = 搬家</span></h3><blockquote style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.55);line-height: 1.8;margin-bottom: 24px;"><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>文件系统与进程提供方共享同一个执行世界,因此把它们指向远程沙箱,也就把 Bash、PTY 和 LSP 一并搬了过去。</span></p></blockquote><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>拆开看:</span></p><ol style="list-style-type: decimal;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><code><span>ctx.fs</span></code><section><span> 有多个 Provider(local/sandbox/e2b)</span></section></li><li style="margin-bottom: 0px;"><code><span>ctx.subprocess</span></code><section><span> 有多个 Provider(local/e2b)</span></section></li><li style="margin-bottom: 0px;"><code><span>ctx.shell</span></code><section><span> 通过 </span><code><span>ctx.subprocess</span></code><span> 执行</span></section></li><li style="margin-bottom: 0px;"><code><span>ctx.lsp</span></code><section><span> 也通过 </span><code><span>ctx.subprocess</span></code><span> 启动</span></section></li></ol><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">所以:把subprocess和fs的 Provider 从 local 换成 e2b,shell、terminal、lsp全部自动跟着去远程。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>7.4 isolate realm</span></h3><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">realm 是比 scope 更严格的服务隔离</span>:scope 是"按 agent 划分视角",realm 是"一个 agent 完全拥有自己的服务实例"。</span></p><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>主 agent 的 ctx.llm 配置 A 模型,子 agent 的 ctx.llm 配置 B 模型——<span style="font-weight: 500;">同名的服务,不同的 realm,各自独立</span>。</span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>7.5 Python 对照:带 seam 的架构升级</span></h3><section class="code-snippet__fix code-snippet__js"><ul class="code-snippet__line-index code-snippet__js"></ul><pre class="code-snippet__js">from abc import ABC, abstractmethod class 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).stdout class 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”))</pre></section><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span><span style="font-weight: 500;">对照 dsh 的差距:</span>dsh 的 Provider 是插件(通过 cordis 配置加载,可热插拔),realm/scope 提供运行时隔离。<span style="font-weight: 500;">但"接口定义 → Provider 注册 → 配置切换 → 业务不变"这条链已跑通。</span></span></p><h3 style="font-size: 17px;font-weight: 400;color: #2B77BF;line-height: 1.8;margin-bottom: 24px;"><span>7.6 本章小结</span></h3><ul style="font-size: 15px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span>seam = <span style="font-weight: 500;">接口声明 + Provider + Consumer</span> 三角色</span></section></li><li style="margin-bottom: 0px;"><code><span>import type</span></code><section><span> 在编译期强制解耦</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">换 Provider = 搬家</span>:fs/subprocess 换 Provider,shell/lsp/terminal 全部跟随</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">isolate realm</span>:agent 拥有自己的服务实例</span></section></li></ul><hr/><h2 style="font-size: 17px;font-weight: 500;color: #2B77BF;line-height: 1.8;margin-bottom: 12px;"><span>中篇小结</span></h2><p style="font-size: 17px;font-weight: 400;color: rgba(0,0,0,0.9);line-height: 1.8;margin-bottom: 24px;"><span>四章下来,你已经掌握了让 Agent 能干活、可替换、可扩展的机制:</span></p><ol style="list-style-type: decimal;" class="list-paddingleft-1"><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">工具系统</span>:schema 驱动、五段流水线、决策与执行分离、并发声明、作用域隔离</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">提示词组装</span>:四路输入、组装/渲染分离、变量后插值、complete 独占</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">Cordis 插件内核</span>:三类事件语义、scope、卸载回滚</span></section></li><li style="margin-bottom: 0px;"><section><span><span style="font-weight: 500;">能力缝 seam</span>:接口定义 + Provider + Consumer,换 Provider = 搬家</span></section></li></ol> <div class="content-mask">
<a href="javascript:void (0);" class="mask-text login-trigger">
<i class="iconfont icon-suo"></i>
登录查看剩余 70% 内容
<i class="iconfont icon-arrow-right-o"></i>
</a>
</div>
</div>
</div>
免费获取企业 AI 成熟度诊断报告,发现转型机会
关注公众号

扫码关注,获取最新 AI 资讯
3 步完成企业诊断,获取专属转型建议
已有 200+ 企业完成诊断