DeepSeek Harness 自动化编排实战：从插件开发到端到端 Agent 流水线

实现一个自定义 LLM 适配器连接云托管模型端点并完成一次流式对话。

上一节你已经看清了一次流式响应的完整生命周期：LLM 适配器按顺序发射分块，消费方按到达顺序把增量追加到界面，结束块标记流结束，错误以结构化方式上报，瞬时错误由适配器通过重试与退避消化。现在把第 4-2 节的 StreamChunk 协议接到真实端点：写一个插件，实现第 4-1 节定义的适配器接口，把它注册到 ctx.llm 的适配器接缝。适配器本身就是插件——模型、工具、会话、Agent 循环全部是可替换插件，没有需要打补丁的特权核心。目标是你公司网关暴露的兼容端点：资料写明，只要网关暴露兼容端点，就能作为自定义提供方接入。你要完成四件事：实现接口、连接端点、把云端响应切成 StreamChunk、处理认证与错误重试。最后在会话里验证流式回复。

实现适配器插件

1
新建插件目录 cloud-adapter/ 与入口文件 index.ts。插件形态在 package.json 的 dsh 字段里声明，也可先作为本地插件直接挂载。

导出插件：apply 接收 (ctx, config)，config 提供 endpoint 与 apiKey。把适配器注册到 ctx.llm 接缝；注册是效应，插件卸载时自动回卷。

typescript

export const name = 'cloud-adapter'
export function apply(ctx: Context, config: { endpoint: string; apiKey: string }) {
  const adapter = { async *stream(messages: Message[]) { /* 步骤 3-5 */ } }
  ctx.effect(() => registerAdapter(ctx.llm, adapter))
}

在 stream 里用 fetch 请求 config.endpoint，请求体带 messages 与 stream: true，请求头带 Authorization: Bearer <apiKey>。非 2xx 抛错。

typescript

const res = await fetch(config.endpoint, {
  method: 'POST',
  headers: { Authorization: `Bearer ${config.apiKey}`, 'Content-Type': 'application/json' },
  body: JSON.stringify({ messages, stream: true }),
})
if (!res.ok) throw new Error(`cloud endpoint ${res.status}`)

4
逐行读取响应体，每行增量按第 4-2 节的 StreamChunk 协议转成分块并立即 yield；读到结束标记后 yield 结束块。不要整段缓冲。
typescript
```
for await (const line of readSse(res.body)) {
  if (isDoneLine(line)) break
  const delta = extractDelta(line)
  if (delta) yield toStreamChunk(delta)
}
yield endChunk()
```

认证错误与瞬时故障分开：401/403 直接失败；429 与 5xx 有界指数退避重试，最多 3 次。

typescript

const MAX_RETRIES = 3
for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
  const res = await fetch(config.endpoint, opts) // 同前
  if (res.status === 401 || res.status === 403) throw new Error('bad credential')
  if (res.status === 429 || res.status >= 500) {
    if (attempt === MAX_RETRIES) throw new Error(`still failing: ${res.status}`)
    await sleep(200 * 2 ** attempt)
    continue
  }
  if (!res.ok) throw new Error(`cloud endpoint ${res.status}`)
  break
}

你应该看到

ctx.llm 上出现你注册的适配器，卸载插件后注册被回卷；插件能通过类型检查并被 dsh 加载。

凭据只走凭据通道

apiKey 是凭据：不要写死在插件源码、patch 文件或提交记录里，也不要把 Authorization 头写进日志。dsh 有独立的凭据子系统（docs 的 /docs/subsystems/credentials），默认 DeepSeek 路由本身就要求一份凭据。只连接你被授权调用的端点。日志只记录状态码与耗时，不记录头与响应体；401/403 不要重试。

写代码前先手动验证端点

写适配器之前，先用 curl 对 config.endpoint 发一次带 Authorization 头的请求，确认响应格式、结束标记与错误状态码。readSse、isDoneLine、extractDelta 三个函数写的就是你观察到的格式；格式不匹配是流式适配最常见的故障点。

装配到 profile 并验证流式对话

1
在 profile 的 cordis.patch.yml 里插入一行挂载插件。patch 按行 id 定位：替换已有行的整行配置，或插入新行。给新行一个稳定 id，配置里放 endpoint 与 apiKey，格式按第 2-1 节。
yaml
```
rows:
  - id: cloud-adapter
    plugin: ./cloud-adapter
    config:
      endpoint: https://gateway.example.com/v1/chat/completions
      # apiKey 从凭据子系统取，不写死
```
2
启动本地 Web UI。
bash
```
npx @deepseek-ai/dsh web
```
3
浏览器打开 http://127.0.0.1:3080。资料建议固定包版本，测试基线为 @deepseek-ai/dsh 0.1.0-rc.6，Node 需 ^22.19.0 或 >=24.0.0。
4
按第 1-2 节的模型提供方配置，把会话默认模型切到你的适配器；API 凭据在 Settings → Models 维护。
5
新建会话，输入一句要求流式输出的提问，观察回复是否逐字出现；打开 Trajectory 视图核对分块序列与结束块。

你应该看到

会话中输出由自定义适配器产生的流式回复；Trajectory 视图显示 delta 分块序列与结束块；错误路径下只见对 429/5xx 的重试，不见对 401 的重试。

遇到问题：模型列表里没有你的适配器，或插件加载报错。

可能原因

patch 行 id 与目标行不一致；patch 按行 id 定位，对不上就替换不到。
patch 放错层级：profile 自带的 cordis.patch.yml 在 bundle 之后、home 级 patch 之前生效。
Node 版本不满足引擎要求（^22.19.0 或 >=24.0.0）。

排查步骤

先打印启动树，看你机器实际挂载的行，再核对行 id。
把新行放进 profile 自己的 cordis.patch.yml。
升级 Node 后重启 dsh。

遇到问题：请求返回 401/403，或会话长时间无输出后一次性给出整段文字。

可能原因

凭据缺失或失效；默认 DeepSeek 路由要求凭据，你的网关同样要求。
认证方式不匹配：服务端要的不是 Bearer。
响应被整段缓冲后才返回，没有按 StreamChunk 逐块产出。

排查步骤

在凭据子系统确认凭据有效，向网关确认认证方式。
非 2xx 必须抛错，不要返回一个成功的空流。
在读取循环里即时 yield 分块，不要等流结束。
只对 429/5xx 重试；401/403 直接失败并给出可读错误。

自检：你的适配器收到 401 时该进重试分支吗？为什么？哪种状态码才值得重试？

不该。401 表示凭据无效或缺失，重试不会改变结果，只会放大请求量；凭据错误应直接失败并提示检查凭据，默认 DeepSeek 路由同样要求凭据。值得重试的是 429 与 5xx 这类瞬时故障，用有界指数退避，并设重试上限。