DeepSeek Harness: LLM 适配器

Q: 自定义适配器需要实现哪些方法？

必须实现 complete 、 stream 、 getModels 、 getModelCapabilities 。缺少任何方法，TypeScript 编译报错。

Q: 如何测试自定义适配器？

用 InMemoryProvider 模式——不发送真实 HTTP 请求，返回预设的响应。单元测试中替换 llm 服务即可。

Q: 流式输出中断了怎么办？

StreamChunk 协议包含 error 类型。消费者应处理错误 chunk，记录日志后决定是否重试。

Q: 适配器中的 API Key 如何安全存储？

使用环境变量或 .env 文件，不要硬编码在配置中。Schemastery 支持 .hidden() 标记敏感字段。 ---

最后更新：2026-08-31

DSH 的"模型无关"不是口号，而是架构设计——LLM 适配器作为 Cordis 能力系统的 Provider，让切换模型像换电池一样简单。本课深入 LLM 适配器的注册、实现和多模型管理。

💡 提示：所有 LLM 适配器都实现同一个 Definition——LLMCapability。Agent 和工具只依赖这个接口，不关心底层是 DeepSeek 还是 GPT-4o。

📋 前置知识：已完成 22-capability.md，理解能力三角色

1. 你将学到

ctx.llm 适配器注册
OpenAI 兼容端点适配器
自定义模型路由
StreamChunk 协议
模型能力声明
多模型负载均衡

2. ctx.llm 适配器注册

LLM 适配器架构

(1) LLM 能力接口

DSH 的 LLM 适配器实现 LLMCapability 接口：

TYPESCRIPT

interface LLMCapability {
  complete(request: LLMRequest): Promise<LLMResponse>
  stream(request: LLMRequest): AsyncIterable<StreamChunk>
  getModels(): ModelInfo[]
  getModelCapabilities(model: string): ModelCapabilities
}

(2) ▶ 示例 2

TYPESCRIPT

import { Service, Context } from '@deepseek-ai/cordis'

export default class MyLLMAdapter extends Service {
  constructor(ctx: Context) {
    super(ctx, 'llm')
  }
}

super(ctx, 'llm') 将适配器注册为 llm 服务。

(3) 内置适配器

DSH 自带两个 LLM 适配器：

适配器	服务名	支持的模型
DeepSeek	`llm`	deepseek-chat, deepseek-reasoner
OpenAI Compatible	`llm`	任何 OpenAI 兼容 API

(4) ▶ 示例 4

YAML

# cordis.yml
plugins:
  llm:
    $replace: ./adapters/my-custom-llm
    config:
      apiKey: sk-xxx
      endpoint: https://api.example.com/v1

3. OpenAI 兼容端点适配器

(1) 协议概述

OpenAI Chat Completions API 是事实上的行业标准。DSH 的内置适配器兼容此协议：

TYPESCRIPT

interface OpenAIRequest {
  model: string
  messages: { role: string; content: string }[]
  temperature?: number
  max_tokens?: number
  stream?: boolean
  tools?: ToolDefinition[]
}

(2) ▶ 示例 2

YAML

# cordis.yml
plugins:
  llm:
    config:
      provider: openai-compatible
      apiKey: sk-xxx
      endpoint: https://api.openai.com/v1
      models:
        - id: gpt-4o
          capabilities: [chat, tool-use, vision]
        - id: gpt-4o-mini
          capabilities: [chat, tool-use]

(3) 自定义端点

任何 OpenAI 兼容的 API 都可以接入：

YAML

# 接入本地 Ollama
plugins:
  llm:
    config:
      provider: openai-compatible
      endpoint: http://localhost:11434/v1
      apiKey: ollama
      models:
        - id: llama3
          capabilities: [chat]

# 接入 Azure OpenAI
plugins:
  llm:
    config:
      provider: openai-compatible
      endpoint: https://my-resource.openai.azure.com/openai/deployments/my-deployment
      apiKey: xxx
      headers:
        api-key: xxx

(4) 编写自定义适配器

当内置适配器不满足需求时，可以编写自定义适配器：

TYPESCRIPT

import { Service, Context } from '@deepseek-ai/cordis'
import { LLMCapability, LLMRequest, LLMResponse, StreamChunk } from '@deepseek-ai/dsh'

export const Config = Schema.object({
  apiKey: Schema.string().required().description('API key'),
  endpoint: Schema.string().required().description('API endpoint'),
  defaultModel: Schema.string().default('custom-model').description('Default model ID')
})

export default class CustomLLMAdapter extends Service implements LLMCapability {
  static inject = []

  private apiKey: string
  private endpoint: string
  private defaultModel: string

  constructor(ctx: Context) {
    super(ctx, 'llm')
    this.apiKey = ctx.config.apiKey
    this.endpoint = ctx.config.endpoint
    this.defaultModel = ctx.config.defaultModel
  }

  async complete(request: LLMRequest): Promise<LLMResponse> {
    const response = await fetch(`${this.endpoint}/chat/completions`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.apiKey}`
      },
      body: JSON.stringify({
        model: request.model || this.defaultModel,
        messages: request.messages,
        temperature: request.temperature,
        max_tokens: request.maxTokens,
        stream: false
      })
    })

    const data = await response.json()
    return {
      content: data.choices[0].message.content,
      model: data.model,
      usage: data.usage
    }
  }

  async *stream(request: LLMRequest): AsyncIterable<StreamChunk> {
    const response = await fetch(`${this.endpoint}/chat/completions`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.apiKey}`
      },
      body: JSON.stringify({
        model: request.model || this.defaultModel,
        messages: request.messages,
        temperature: request.temperature,
        max_tokens: request.maxTokens,
        stream: true
      })
    })

    const reader = response.body!.getReader()
    const decoder = new TextDecoder()
    let buffer = ''

    while (true) {
      const { done, value } = await reader.read()
      if (done) break

      buffer += decoder.decode(value, { stream: true })
      const lines = buffer.split('\n')
      buffer = lines.pop() || ''

      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const data = line.slice(6)
          if (data === '[DONE]') return
          try {
            const parsed = JSON.parse(data)
            const delta = parsed.choices[0].delta
            if (delta.content) {
              yield { type: 'text', content: delta.content }
            }
          } catch {}
        }
      }
    }
  }

  getModels() {
    return [{ id: this.defaultModel, capabilities: ['chat', 'tool-use'] }]
  }

  getModelCapabilities(model: string) {
    return { chat: true, toolUse: true, vision: false }
  }
}

4. 自定义模型路由

(1) 路由概念

模型路由根据请求特征选择不同的 LLM 适配器：

graph TB
    REQ[LLM 请求] --> ROUTER{模型路由}
    ROUTER -->|code 任务| CODE[DeepSeek Coder]
    ROUTER -->|reasoning| REASON[DeepSeek Reasoner]
    ROUTER -->|简单对话| CHAT[GPT-4o-mini]

(2) 配置路由

YAML

plugins:
  llm:
    config:
      routing:
        default: deepseek-chat
        rules:
          - match:
              mode: ptc
            model: deepseek-reasoner
          - match:
              mode: creative
            model: deepseek-chat
          - match:
              tools: [shell]
            model: deepseek-coder

(3) 自定义路由逻辑

TYPESCRIPT

export default class RoutingLLMAdapter extends Service {
  private adapters: Map<string, LLMCapability> = new Map()

  constructor(ctx: Context) {
    super(ctx, 'llm')
  }

  async complete(request: LLMRequest): Promise<LLMResponse> {
    const model = this._selectModel(request)
    const adapter = this._getAdapter(model)
    return adapter.complete({ ...request, model })
  }

  private _selectModel(request: LLMRequest): string {
    if (request.tools?.some(t => t.name === 'shell')) {
      return 'deepseek-coder'
    }
    if (request.metadata?.mode === 'ptc') {
      return 'deepseek-reasoner'
    }
    return 'deepseek-chat'
  }

  private _getAdapter(model: string): LLMCapability {
    const prefix = model.split('-')[0]
    return this.adapters.get(prefix) || this.adapters.get('default')!
  }
}

5. StreamChunk 协议

(1) 协议定义

StreamChunk 是 DSH 的流式输出统一协议：

TYPESCRIPT

type StreamChunk =
  | { type: 'text'; content: string }
  | { type: 'tool_call'; id: string; name: string; arguments: string }
  | { type: 'tool_result'; id: string; result: any }
  | { type: 'error'; error: Error }
  | { type: 'done'; reason: 'stop' | 'tool_use' | 'length' }

(2) chunk 类型说明

类型	说明	来源
`text`	文本内容片段	LLM 流式输出
`tool_call`	工具调用请求	LLM 决定调用工具
`tool_result`	工具执行结果	工具执行后
`error`	错误信息	任何阶段的错误
`done`	流结束	LLM 输出结束

(3) 消费 StreamChunk

TYPESCRIPT

for await (const chunk of ctx.llm.stream(request)) {
  switch (chunk.type) {
    case 'text':
      process.stdout.write(chunk.content)
      break
    case 'tool_call':
      console.log(`\n调用工具: ${chunk.name}`)
      break
    case 'tool_result':
      console.log(`工具结果:`, chunk.result)
      break
    case 'error':
      console.error(`错误:`, chunk.error)
      break
    case 'done':
      console.log(`\n完成 (${chunk.reason})`)
      break
  }
}

6. 模型能力声明

(1) ModelCapabilities 接口

TYPESCRIPT

interface ModelCapabilities {
  chat: boolean
  toolUse: boolean
  vision: boolean
  maxTokens: number
  supportedModes: string[]
}

(2) 声明模型能力

TYPESCRIPT

getModels(): ModelInfo[] {
  return [
    {
      id: 'deepseek-chat',
      capabilities: {
        chat: true,
        toolUse: true,
        vision: false,
        maxTokens: 65536,
        supportedModes: ['standard', 'ptc', 'minimal', 'creative']
      }
    },
    {
      id: 'deepseek-reasoner',
      capabilities: {
        chat: true,
        toolUse: true,
        vision: false,
        maxTokens: 65536,
        supportedModes: ['ptc']
      }
    }
  ]
}

(3) 能力声明的作用

框架根据能力声明决定：

是否允许该模型使用工具（toolUse）
是否允许发送图片（vision）
最大 token 限制（maxTokens）
支持的运行模式（supportedModes）

7. 多模型负载均衡

(1) 配置

YAML

plugins:
  llm:
    config:
      loadBalancing:
        strategy: round-robin
        endpoints:
          - model: deepseek-chat
            endpoint: https://api.deepseek.com/v1
            apiKey: sk-key1
            weight: 3
          - model: gpt-4o
            endpoint: https://api.openai.com/v1
            apiKey: sk-key2
            weight: 1

(2) 负载均衡策略

策略	说明
`round-robin`	轮询
`random`	随机选择
`weighted`	按权重分配
`least-latency`	选择延迟最低的

(3) 自定义负载均衡

TYPESCRIPT

export default class LoadBalancedLLM extends Service {
  private endpoints: Endpoint[]
  private currentIndex = 0

  async complete(request: LLMRequest): Promise<LLMResponse> {
    const endpoint = this._nextEndpoint()
    return endpoint.complete(request)
  }

  private _nextEndpoint(): Endpoint {
    const endpoint = this.endpoints[this.currentIndex]
    this.currentIndex = (this.currentIndex + 1) % this.endpoints.length
    return endpoint
  }
}

❓ 常见问题

Q 可以同时注册多个 LLM 适配器吗？

A 默认不行，后注册的覆盖先注册的。用 realm 隔离或路由适配器来管理多模型。

Q 自定义适配器需要实现哪些方法？

A 必须实现 complete、stream、getModels、getModelCapabilities。缺少任何方法，TypeScript 编译报错。

Q 如何测试自定义适配器？

A 用 InMemoryProvider 模式——不发送真实 HTTP 请求，返回预设的响应。单元测试中替换 llm 服务即可。

Q 流式输出中断了怎么办？

A StreamChunk 协议包含 error 类型。消费者应处理错误 chunk，记录日志后决定是否重试。

Q 模型路由和负载均衡有什么区别？

A 路由是根据请求特征选择模型（代码→coder），负载均衡是在同模型的多端点间分配请求。

Q 适配器中的 API Key 如何安全存储？

A 使用环境变量或 .env 文件，不要硬编码在配置中。Schemastery 支持 .hidden() 标记敏感字段。 ---

📖 小节

LLM 适配器是 Cordis 能力系统的 Provider，实现 LLMCapability 接口
内置 OpenAI 兼容适配器，支持任何兼容 API（DeepSeek、GPT-4o、Ollama 等）
自定义适配器实现 complete/stream/getModels/getModelCapabilities 四个方法
模型路由根据请求特征选择不同模型，负载均衡在同模型多端点间分配
StreamChunk 协议统一流式输出：text/tool_call/tool_result/error/done
模型能力声明告诉框架模型能做什么，框架据此限制功能

📝 作业

1. ⭐ 基础题：配置 DSH 使用 OpenAI 兼容端点接入 GPT-4o，设置 API Key 和模型 ID，让 Agent 成功完成一次对话。

2. ⭐⭐ 进阶题：编写一个自定义 LLM 适配器，将请求转发到本地 Ollama 服务（http://localhost:11434/v1）。实现 complete 和 stream 方法，测试流式输出。

3. ⭐⭐⭐ 挑战题：实现一个路由适配器，根据 Agent 运行模式选择模型——PTC 模式用 reasoner，其他模式用 chat。创建两个会话分别使用不同模式，验证它们使用了不同的模型。

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

DeepSeek Harness: LLM 适配器

1. 你将学到

2. ctx.llm 适配器注册

(1) LLM 能力接口

(2) ▶ 示例 2

(3) 内置适配器

(4) ▶ 示例 4

3. OpenAI 兼容端点适配器

(1) 协议概述

(2) ▶ 示例 2

(3) 自定义端点

(4) 编写自定义适配器

4. 自定义模型路由

(1) 路由概念

(2) 配置路由

(3) 自定义路由逻辑

5. StreamChunk 协议

(1) 协议定义

(2) chunk 类型说明

(3) 消费 StreamChunk

6. 模型能力声明

(1) ModelCapabilities 接口

(2) 声明模型能力

(3) 能力声明的作用

7. 多模型负载均衡

(1) 配置

(2) 负载均衡策略

(3) 自定义负载均衡

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好