DeepSeek Harness：开源智能体框架的架构革命，从 Cordis 插件驱动到 Vibe Coding 全链路实战

2026年8月14日，DeepSeek 正式开源了智能体运行框架 DeepSeek Harness（DSH）。发布仅约1.5小时，GitHub Star 突破2.4万，刷新了 GitHub 史上最快涨星纪录。这个数字是什么概念？DeepSeek R1 破2万星用了5.7天，xAI 的 Grok-1 用了1.2天，而 DSH 只用了不到两小时。

但 Star 数量从来不是衡量技术价值的标尺。本文从源码出发，深度拆解 DSH 的架构设计：Cordis 插件驱动模型、统一上下文管理层、多工具编排引擎，以及如何用它从零构建一个具备代码执行、文件操作和网页搜索能力的生产级智能体。整个过程你会看到，DSH 不仅仅是一个"调用工具的包装器"，而是一套面向 AI Native 时代的运行时基础设施。

一、为什么需要 DeepSeek Harness？

在 DSH 出现之前，我们已经有 LangChain、LangGraph、AutoGen、CrewAI 等一大票 Agent 框架。它们解决了一部分问题，但也带来了新的混乱：

工具调用接口不统一。 每个框架定义自己的 Tool 接口，Python 生态里光 langchain-core 的 BaseTool 就分了三代。如果你想在两个框架之间迁移代码，工具层几乎要重写。

上下文管理碎片化。 长对话的上下文怎么管理？历史消息怎么截断？工具调用结果怎么注入？大多数框架把这些问题甩给用户自己处理。

执行环境隔离不足。 一个 Agent 操作本地文件系统、执行 shell 命令、调用 API——这些操作如果缺乏沙箱隔离，生产环境就是一颗定时炸弹。

DSH 的设计目标直指这三个痛点：

1. 统一插件协议：一切皆插件，Cordis 接口规范定义工具、数据源、执行环境的行为边界。
2. 统一上下文管理层：框架内置 History、Memory、Context Window 的管理策略。
3. 安全执行环境：插件运行在隔离的运行时中，权限可精确控制。

二、Cordis 插件协议：一切皆插件

2.1 什么是 Cordis？

Cordis 是 DSH 的核心抽象层。名字源自拉丁语 cor（心脏）和 dis（分开），意为"从心脏延伸出去的分支"——这正好对应了它的设计哲学：Agent 是心脏，插件是延伸到外部世界的触手。

Cordis 协议定义了三类插件：

每一类插件都实现了统一的接口规范。以 Tool 插件为例：

// Cordis Tool 插件接口（TypeScript 定义，实际为 Go 实现）
interface CordisTool {
  // 插件唯一标识
  name: string;
  // 人类可读的描述，LLM 用这个来决定是否调用
  description: string;
  // 参数模式定义（JSON Schema）
  parameters: {
    type: 'object';
    properties: Record<string, SchemaField>;
    required?: string[];
  };
  // 核心执行方法
  execute(params: Record<string, unknown>, ctx: ExecutionContext): Promise<ToolResult>;
  // 可选：执行前验证
  validate?(params: Record<string, unknown>): ValidationResult;
}

// 执行上下文：包含认证信息、权限范围、当前工作目录等
interface ExecutionContext {
  credentials: Record<string, string>;    // API Key 等认证信息
  permissions: PermissionScope;          // 权限范围（文件系统、进程、网络）
  workingDir: string;                    // 当前工作目录
  sessionId: string;                     // 会话 ID，用于上下文追踪
}

这套接口的精妙之处在于：描述即契约。description 和 parameters 字段直接暴露给 LLM，LLM 根据这些信息自主决定调用哪个工具、传什么参数。框架本身不预设任何工具调用策略。

2.2 内置插件体系

DSH 内置了一套开箱即用的插件生态，覆盖了大多数日常开发场景：

// DSH 内置插件一览
plugins/
├── filesystem/        // 文件系统操作
│   ├── read.go       // 读取文件，支持通配符和行号范围
│   ├── write.go      // 写入文件，自动创建目录
│   ├── glob.go       // 文件模式匹配
│   └── tree.go       // 目录树展示
├── terminal/         // 终端执行
│   ├── bash.go       // 执行 Bash 命令，超时控制
│   └── script.go     // 执行脚本文件
├── web/              // 网络操作
│   ├── fetch.go      // HTTP GET/POST，支持自定义 Header
│   ├── search.go     // 网页搜索（内置 DuckDuckGo）
│   └── scrape.go     // 网页内容抓取
├── code/             // 代码相关
│   ├── interpreter.go // Python/JS 代码执行（沙箱隔离）
│   └── linter.go     // 代码检查
└── agent/            // 子 Agent
    ├── planner.go    // 任务规划子 Agent
    ├── reviewer.go   // 代码审查子 Agent
    └── researcher.go // 研究调研子 Agent

每个插件都实现了完整的 Cordis 接口。让我们看一个具体的例子——文件系统读取插件：

// plugins/filesystem/read.go（简化版核心逻辑）
package filesystem

import (
    "context"
    "fmt"
    "os"
    "path/filepath"
    "strconv"
    "strings"

    "github.com/deepseek-ai/harness/cordis"
)

// ReadTool 实现了 cordis.Tool 接口
type ReadTool struct {
    basePath string // 限制只能在特定目录下读取（安全边界）
}

func (t *ReadTool) Name() string { return "read_file" }
func (t *ReadTool) Description() string {
    return `读取文件内容。支持以下参数：
- path: 文件路径（相对于工作目录）
- start_line: 开始行号（可选，从1开始）
- end_line: 结束行号（可选）
- show_lines: 是否显示行号（可选，默认false）`
}
func (t *ReadTool) Parameters() cordis.Schema {
    return cordis.Schema{
        Type: "object",
        Properties: map[string]cordis.SchemaField{
            "path":        {Type: "string", Description: "文件路径"},
            "start_line":  {Type: "integer", Description: "开始行号", Default: 1},
            "end_line":    {Type: "integer", Description: "结束行号"},
            "show_lines":  {Type: "boolean", Description: "显示行号", Default: false},
        },
        Required: []string{"path"},
    }
}

func (t *ReadTool) Execute(ctx context.Context, params map[string]interface{}) (cordis.ToolResult, error) {
    // 安全检查：禁止路径穿越
    requestedPath := params["path"].(string)
    absPath, err := filepath.Abs(filepath.Join(t.basePath, requestedPath))
    if err != nil {
        return cordis.ToolResult{Success: false, Error: err.Error()}, nil
    }
    if !strings.HasPrefix(absPath, t.basePath) {
        return cordis.ToolResult{
            Success: false,
            Error:   "访问被拒绝：路径超出允许范围",
        }, nil
    }

    // 行范围处理
    startLine := 1
    if sl, ok := params["start_line"].(float64); ok {
        startLine = int(sl)
    }
    showLines := false
    if sl, ok := params["show_lines"].(bool); ok {
        showLines = sl
    }

    // 读取文件
    content, err := os.ReadFile(absPath)
    if err != nil {
        return cordis.ToolResult{Success: false, Error: err.Error()}, nil
    }

    lines := strings.Split(string(content), "\n")
    if startLine > len(lines) {
        return cordis.ToolResult{Success: false, Error: "开始行号超出文件范围"}, nil
    }

    endLine := len(lines)
    if el, ok := params["end_line"].(float64); ok {
        endLine = int(el)
    }

    resultLines := lines[startLine-1 : min(endLine, len(lines))]
    var output strings.Builder
    for i, line := range resultLines {
        if showLines {
            output.WriteString(fmt.Sprintf("%4d | %s\n", startLine+i, line))
        } else {
            output.WriteString(line + "\n")
        }
    }

    return cordis.ToolResult{
        Success: true,
        Output:   output.String(),
        Metadata: map[string]interface{}{
            "file_size": len(content),
            "lines_read": len(resultLines),
            "total_lines": len(lines),
        },
    }, nil
}

这段代码展示了 DSH 插件设计的几个核心原则：

1. 安全边界内置：通过 basePath 限制文件操作范围，禁止路径穿越攻击。
2. Schema 自描述：参数定义直接暴露给 LLM，无需额外文档。
3. 结构化输出：ToolResult 包含 Success、Output、Error、Metadata，LLM 可以精确判断工具是否成功。
4. 元数据丰富：返回文件大小、行数等额外信息，帮助 LLM 做出更好的后续决策。

2.3 插件注册与发现

DSH 使用声明式插件注册机制。在项目根目录的 harness.yaml 中定义插件：

# harness.yaml
version: "1.0"

agent:
  model: "deepseek/deepseek-chat-v3"
  temperature: 0.7
  max_tokens: 8192

plugins:
  # 内置插件：文件系统操作
  - name: filesystem
    type: plugin_group
    config:
      base_path: "./workspace"      # 安全边界：只能操作这个目录
      max_file_size: "10MB"

  # 内置插件：终端执行
  - name: terminal
    type: plugin_group
    config:
      allowed_commands:           # 白名单：只允许这些命令
        - "git"
        - "npm"
        - "python3"
        - "go"
      timeout_seconds: 30
      max_output_lines: 500

  # 内置插件：网络操作
  - name: web
    type: plugin_group
    config:
      search_engine: "duckduckgo"
      max_fetch_size: "1MB"
      user_agent: "DSH-Agent/1.0"

  # 自定义插件：从 Python 包索引查询依赖
  - name: pypi_lookup
    type: tool
    class: "custom.pypi.PyPITool"
    config:
      cache_ttl_seconds: 3600

启动时 DSH 解析这个配置文件，自动注册所有插件到 Cordis 总线。LLM 通过 /tools 接口动态获取可用工具列表——这个过程完全透明，开发者无需在代码里硬编码。

三、统一上下文管理层：让长对话不再失控

3.1 上下文管理的核心挑战

当 Agent 需要处理复杂、长时的任务时，上下文管理成为决定成败的关键。常见的问题包括：

• 上下文窗口溢出：输入 token 超过模型限制，任务被迫中断。
• 历史信息噪音：早期的对话内容可能与当前任务无关，但占据了大量 token 预算。
• 工具调用结果碎片化：每次工具调用返回的数据格式不一致，难以整合。
• 多轮推理状态丢失：Agent 在长任务中需要记住中间结论，但上下文窗口有限。

DSH 的上下文管理层（Context Manager）设计了一套完整的解决方案。

3.2 分层记忆架构

DSH 将 Agent 的记忆划分为三个层次，每层有不同的淘汰策略：

┌─────────────────────────────────────────┐
│          Working Context (热存储)         │  ← 当前任务相关，直接参与推理
│    最近 N 条消息 + 最新工具调用结果          │     淘汰策略：精确裁剪（Tail pruning）
├─────────────────────────────────────────┤
│         Episodic Memory (温存储)          │  ← 任务片段记忆，按重要性评分
│    重要决策、关键结论、用户偏好              │     淘汰策略：重要性评分 (Recency + Relevance)
├─────────────────────────────────────────┤
│          Semantic Memory (冷存储)         │  ← 长期知识，可按需召回
│    领域知识、项目背景、API 文档摘要          │     淘汰策略：向量检索 (Vector RAG)
└─────────────────────────────────────────┘

关键代码实现：

// context/manager.go
type ContextManager struct {
    working      *WorkingMemory
    episodic     *EpisodicMemory
    semantic     *SemanticMemory

    maxTokens    int                    // 最大 token 预算
    currentUsed  int                    // 当前已使用 token 数

    // 淘汰策略配置
    pruneConfig  PruneConfig
}

// WorkingMemory：精确裁剪，保留最近的消息和最新结果
func (wm *WorkingMemory) Prune(targetTokens int) {
    for wm.tokenCount() > targetTokens {
        // 从最老的消息开始裁剪
        oldest := wm.messages[0]
        wm.messages = wm.messages[1:]
        wm.tokenCount.Subtract(countTokens(oldest))

        // 如果该消息有工具调用结果，也一并裁剪
        if oldest.ToolResults != nil {
            for _, tr := range oldest.ToolResults {
                wm.tokenCount.Subtract(countTokens(tr))
            }
        }
    }
}

// EpisodicMemory：重要性评分淘汰
type EpisodicMemory struct {
    episodes   []Episode
    importance *ImportanceScorer  // 基于关键词频率 + 位置 + 工具调用评分
}

type Episode struct {
    Content    string
    Score      float64             // 重要性评分 0-1
    CreatedAt  time.Time
    Metadata   map[string]string
}

func (em *EpisodicMemory) Prune(targetTokens int) {
    // 评分低于阈值的 episode 先淘汰
    threshold := em.calculateDynamicThreshold()
    for em.tokenCount() > targetTokens {
        // 找到评分最低且最老的 episode
        minIdx := -1
        for i := range em.episodes {
            if em.episodes[i].Score < threshold {
                if minIdx == -1 || em.episodes[i].CreatedAt.Before(em.episodes[minIdx].CreatedAt) {
                    minIdx = i
                }
            }
        }
        if minIdx == -1 {
            break // 所有 episode 都达到阈值，降低阈值重试
        }
        em.remove(minIdx)
    }
}

// SemanticMemory：向量检索召回
type SemanticMemory struct {
    index   *vector.Index    // FAISS 或 Qdrant 等向量索引
    docs    []Document
}

func (sm *SemanticMemory) Retrieve(query string, topK int) []Document {
    queryVec := sm.embed(query)
    return sm.index.Search(queryVec, topK)
}

这套设计的精妙之处在于分层联动：当 Working Memory 裁剪时，被淘汰的消息不是直接丢弃，而是进入 Episodic Memory 进行重要性评分。评分高的进入 Semantic Memory 存档，下次任务可以通过向量检索召回。

3.3 上下文注入策略

DSH 没有使用简单的"把所有记忆塞进 context"的暴力方案，而是实现了智能上下文注入：

// context/injector.go
type ContextInjector struct {
    manager     *ContextManager
    llm         LLMClient
}

// 构建发送给 LLM 的完整上下文
func (ci *ContextInjector) BuildPrompt(task string) (string, error) {
    var sections []string

    // 1. 系统提示（固定不变）
    sections = append(sections, ci.systemPrompt)

    // 2. 召回的语义记忆（与当前任务相关的知识）
    semanticContext := ci.manager.SemanticRecall(task)
    if len(semanticContext) > 0 {
        sections = append(sections, "📚 相关知识：\n"+semanticContext)
    }

    // 3. 重要片段记忆（高评分的决策和结论）
    episodicContext := ci.manager.EpisodicRecall(task)
    if len(episodicContext) > 0 {
        sections = append(sections, "💡 前期进展：\n"+episodicContext)
    }

    // 4. 当前工作上下文（热存储）
    workingContext := ci.manager.WorkingGetAll()
    sections = append(sections, "📋 当前状态：\n"+workingContext)

    // 5. 当前任务
    sections = append(sections, "🎯 任务：\n"+task)

    // 6. 工具描述（让 LLM 知道可以用什么）
    toolsDesc := ci.manager.GetAvailableToolsDescription()
    sections = append(sections, "🔧 可用工具：\n"+toolsDesc)

    // 组装并检查 token 预算
    prompt := strings.Join(sections, "\n\n")
    if countTokens(prompt) > ci.manager.MaxTokens {
        // 超过预算，触发渐进式缩减
        prompt = ci.manager.ProgressiveCompress(prompt, ci.manager.MaxTokens)
    }

    return prompt, nil
}

这种策略的效果：在 8K token 的上下文窗口限制下，DSH 能有效利用的有效信息密度比直接塞满上下文高出 3-5 倍。

四、多工具编排引擎：让 Agent 做复杂决策

4.1 ReAct + Plan 的混合推理模式

DSH 默认采用 ReAct（Reasoning + Acting） 模式，但针对复杂任务提供了 Plan 模式作为补充：

// engine/agent.go
type Agent struct {
    model       LLMClient
    tools       []cordis.Tool
    context     *ContextManager
    mode        AgentMode  // ReAct | Plan | Auto
}

// ReAct 模式：边想边做，单轮决策
func (a *Agent) RunReAct(ctx context.Context, task string) (*AgentResult, error) {
    for step := 0; step < a.maxSteps; step++ {
        // 1. 构建上下文
        prompt := a.context.BuildPrompt(task)

        // 2. LLM 推理并决定下一步行动
        decision, err := a.model.Decide(prompt)
        if err != nil {
            return nil, err
        }

        // 3. 解析决策：可能是调用工具、回答问题、或请求澄清
        switch decision.Type {
        case ActionCall:
            result := a.executeTool(ctx, decision.ToolName, decision.Params)
            a.context.AddToolResult(decision.ToolName, result)

        case FinalAnswer:
            return &AgentResult{Answer: decision.Answer, Steps: step + 1}, nil

        case NeedClarification:
            return &AgentResult{NeedsClarification: true, Question: decision.Question}, nil
        }
    }
    return nil, errors.New("超过最大步数限制")
}

// Plan 模式：先规划后执行，适合复杂多步骤任务
func (a *Agent) RunPlan(ctx context.Context, task string) (*AgentResult, error) {
    // 阶段一：规划
    planPrompt := a.context.BuildPrompt(fmt.Sprintf(
        `请为以下任务制定执行计划：

任务：%s

要求：
1. 将任务分解为明确的步骤
2. 每一步用一句简洁的话描述目标
3. 标注每一步的预期输出
4. 注意步骤之间的依赖关系

请用以下 JSON 格式输出计划：
{
  "steps": [
    {"id": 1, "description": "...", "expected_output": "...", "depends_on": []}
  ]
}`, task))

    planResp, err := a.model.Generate(planPrompt)
    if err != nil {
        return nil, err
    }

    plan := parsePlan(planResp)
    a.context.AddPlan(plan) // 规划存入上下文，供后续使用

    // 阶段二：逐步执行
    for _, step := range plan.Steps {
        // 构建带规划上下文的提示
        stepPrompt := a.context.BuildPrompt(fmt.Sprintf(
            "当前计划步骤：%d/%d\n目标：%s\n预期输出：%s\n\n请执行此步骤：",
            step.ID, len(plan.Steps), step.Description, step.ExpectedOutput))

        result, err := a.executeStep(ctx, stepPrompt)
        if err != nil {
            // 失败时尝试调整后续计划
            adjustedPlan := a.replan(ctx, plan, step, err)
            if adjustedPlan == nil {
                return nil, fmt.Errorf("步骤 %d 失败且无法重新规划: %w", step.ID, err)
            }
            plan = adjustedPlan
        } else {
            a.context.AddStepResult(step.ID, result)
        }
    }

    // 阶段三：汇总结果
    return a.summarize(plan), nil
}

什么时候用 ReAct，什么时候用 Plan？ DSH 提供了一个启发式规则：

func autoSelectMode(taskComplexity int) AgentMode {
    if taskComplexity < 3 {
        return ReAct  // 简单任务：2-3步搞定，直接做
    } else if taskComplexity < 8 {
        return Auto   // 中等任务：让 LLM 自己决定
    } else {
        return Plan   // 复杂任务：先规划后执行，减少中途迷失
    }
}

4.2 工具调用结果的结构化处理

LLM 工具调用最怕的就是"返回了一堆文本但不知道哪个是关键信息"。DSH 要求每个工具必须返回结构化结果，并在引擎层面做统一处理：

// engine/result_aggregator.go
type ResultAggregator struct {
    schemaValidator *jsonschema.Validator
}

type ToolResult struct {
    Success   bool                   `json:"success"`
    Output    string                 `json:"output,omitempty"`
    Error     string                 `json:"error,omitempty"`
    Metadata  map[string]interface{} `json:"metadata,omitempty"`
    // 关键字段提取（工具可选择性提供）
    KeyFields map[string]interface{} `json:"key_fields,omitempty"`
}

// 聚合多工具结果，统一格式
func (ra *ResultAggregator) Aggregate(results []ToolResult) string {
    var sb strings.Builder
    for _, r := range results {
        if r.Success {
            sb.WriteString(fmt.Sprintf("✅ %s\n", r.Output))
            if len(r.KeyFields) > 0 {
                // 提取关键字段，用更紧凑的格式展示
                sb.WriteString("   关键信息：")
                for k, v := range r.KeyFields {
                    sb.WriteString(fmt.Sprintf(" %s=%v;", k, v))
                }
                sb.WriteString("\n")
            }
        } else {
            sb.WriteString(fmt.Sprintf("❌ %s\n", r.Error))
        }
    }
    return sb.String()
}

五、从零构建一个生产级代码审查 Agent

光说不练假把式。下面我们用 DSH 从零构建一个 代码审查 Agent，它能：

1. 读取代码文件
2. 执行静态分析工具
3. 分析代码复杂度
4. 生成审查报告

5.1 项目初始化

# 安装 DSH CLI
pip install dsh-cli

# 创建新项目
dsh init code-review-agent
cd code-review-agent

# 目录结构
code-review-agent/
├── harness.yaml           # 配置文件
├── plugins/               # 自定义插件
│   └── static_analyzer/
│       ├── __init__.py
│       ├── analyzer.py
│       └── complexity.py
└── prompts/
    ├── system.md
    └── review_template.md

5.2 配置 harness.yaml

version: "1.0"

agent:
  model: "deepseek/deepseek-chat-v3"
  temperature: 0.3          # 代码审查需要稳定性，低温度
  max_tokens: 16384

plugins:
  # 内置：文件系统操作
  - name: filesystem
    type: plugin_group
    config:
      base_path: "./target_repo"    # 审查目标仓库
      max_file_size: "1MB"

  # 内置：终端执行（运行静态分析工具）
  - name: terminal
    type: plugin_group
    config:
      allowed_commands:
        - "python3"
        - "npm"
        - "ruff"
        - "mypy"
        - "eslint"
      timeout_seconds: 60
      working_dir: "./target_repo"

  # 自定义：代码复杂度分析
  - name: complexity_analyzer
    type: tool
    class: "plugins.static_analyzer.ComplexityAnalyzer"
    config:
      cyclomatic_threshold: 10
      cognitive_threshold: 15
      max_lines_per_function: 50

context:
  max_tokens: 12000
  working_memory_lines: 20       # 保留最近20条消息
  importance_threshold: 0.3     # 重要性低于0.3的episode会被淘汰

behavior:
  max_steps: 50                  # 最多50步，防止无限循环
  step_timeout_seconds: 120
  retry_on_failure: true
  retry_count: 2

5.3 自定义复杂度分析插件

这是核心亮点——我们写一个真正的复杂度分析工具，而不是简单调用现成的 CLI：

# plugins/static_analyzer/complexity.py
import ast
import re
from typing import Dict, List, Any
from dataclasses import dataclass
from dsh import CordisTool, ToolResult, ExecutionContext


@dataclass
class FunctionMetrics:
    name: str
    line_start: int
    line_end: int
    cyclomatic_complexity: int
    cognitive_complexity: int
    params_count: int
    return_statements: int
    nested_depth: int


class ComplexityAnalyzer(CordisTool):
    """代码复杂度分析工具"""

    name = "analyze_complexity"
    description = """分析 Python 文件的函数级别复杂度指标。

    返回每个函数的：
    - 圈复杂度（Cyclomatic Complexity）：衡量程序逻辑分支数量
    - 认知复杂度（Cognitive Complexity）：衡量代码理解的难易程度
    - 嵌套深度：最大嵌套层数
    - 参数数量、返回语句数量

    建议阈值：
    - 圈复杂度 > 10 表示函数需要重构
    - 认知复杂度 > 15 表示代码难以理解
    - 嵌套深度 > 4 表示需要提取子函数
    """

    parameters = {
        "type": "object",
        "properties": {
            "file_path": {
                "type": "string",
                "description": "Python 文件路径（相对于工作目录）"
            },
            "show_warnings": {
                "type": "boolean",
                "description": "是否只显示警告项（超过阈值的）",
                "default": False
            }
        },
        "required": ["file_path"]
    }

    def __init__(self, config: Dict[str, Any]):
        self.cyclomatic_threshold = config.get("cyclomatic_threshold", 10)
        self.cognitive_threshold = config.get("cognitive_threshold", 15)
        self.max_lines = config.get("max_lines_per_function", 50)

    def execute(
        self,
        params: Dict[str, Any],
        ctx: ExecutionContext
    ) -> ToolResult:
        file_path = params["file_path"]
        show_warnings = params.get("show_warnings", False)

        try:
            # 读取文件内容
            with open(ctx.full_path(file_path), "r", encoding="utf-8") as f:
                source = f.read()

            # 解析 AST
            tree = ast.parse(source)
            functions = self._extract_functions(tree, source)

            # 生成报告
            if show_warnings:
                functions = [f for f in functions if self._is_warning(f)]

            report = self._generate_report(functions, source)
            return ToolResult(
                success=True,
                output=report,
                metadata={
                    "file": file_path,
                    "total_functions": len(functions),
                    "warning_count": len([f for f in functions if self._is_warning(f)]),
                    "warning_rate": self._warning_rate(functions)
                },
                key_fields={
                    "total_functions": len(functions),
                    "high_complexity": len([f for f in functions if f.cyclomatic_complexity > self.cyclomatic_threshold]),
                    "warning_rate": f"{self._warning_rate(functions):.1f}%"
                }
            )
        except Exception as e:
            return ToolResult(success=False, error=str(e))

    def _extract_functions(self, tree: ast.AST, source: str) -> List[FunctionMetrics]:
        functions = []
        lines = source.split("\n")

        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                metrics = FunctionMetrics(
                    name=node.name,
                    line_start=node.lineno,
                    line_end=node.end_lineno or node.lineno,
                    cyclomatic_complexity=self._cyclomatic_complexity(node),
                    cognitive_complexity=self._cognitive_complexity(node),
                    params_count=len(node.args.args),
                    return_statements=len([n for n in ast.walk(node) if isinstance(n, ast.Return)]),
                    nested_depth=self._max_nesting(node)
                )
                functions.append(metrics)

        return sorted(functions, key=lambda f: f.cyclomatic_complexity, reverse=True)

    def _cyclomatic_complexity(self, node: ast.FunctionDef) -> int:
        """圈复杂度 = E - N + 2P，其中 E=边数，N=节点数，P=连通分量"""
        # 简化计算：分支语句数量 + 1
        complexity = 1
        for child in ast.walk(node):
            if isinstance(child, (ast.If, ast.For, ast.While, ast.ExceptHandler)):
                complexity += 1
            elif isinstance(child, ast.BoolOp):  # and/or
                complexity += len(child.values) - 1
            elif isinstance(child, (ast.Continue, ast.Break)):
                complexity += 1
        return complexity

    def _cognitive_complexity(self, node: ast.FunctionDef) -> int:
        """认知复杂度（简化版）"""
        return self._recursive_cognitive(node, 0)

    def _recursive_cognitive(self, node: ast.AST, depth: int) -> int:
        score = 0
        for child in ast.iter_child_nodes(node):
            if isinstance(child, (ast.If, ast.For, ast.While, ast.ExceptHandler,
                                  ast.With, ast.Match)):
                score += 1 + depth
            elif isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
                # 函数定义增加递归惩罚
                score += depth
            score += self._recursive_cognitive(child, depth + 1)
        return score

    def _max_nesting(self, node: ast.AST) -> int:
        """计算最大嵌套深度"""
        max_depth = [0]

        def visitor(n, depth):
            if isinstance(n, (ast.If, ast.For, ast.While, ast.With)):
                if depth + 1 > max_depth[0]:
                    max_depth[0] = depth + 1
            for child in ast.iter_child_nodes(n):
                visitor(child, depth + (1 if isinstance(n, (ast.If, ast.For, ast.While, ast.With)) else 0))

        visitor(node, 0)
        return max_depth[0]

    def _is_warning(self, f: FunctionMetrics) -> bool:
        return (f.cyclomatic_complexity > self.cyclomatic_threshold or
                f.cognitive_complexity > self.cognitive_threshold or
                (f.line_end - f.line_start) > self.max_lines)

    def _warning_rate(self, functions: List[FunctionMetrics]) -> float:
        if not functions:
            return 0.0
        warnings = len([f for f in functions if self._is_warning(f)])
        return warnings / len(functions) * 100

    def _generate_report(self, functions: List[FunctionMetrics], source: str) -> str:
        lines = source.split("\n")
        var sb = []

        sb.append(f"\n{'='*60}")
        sb.append(f"{'函数':<30} {'复杂度':>8} {'认知':>8} {'深度':>5} {'行数':>5} {'警告':<5}")
        sb.append(f"{'='*60}")

        for f in functions:
            warning = "⚠️" if self._is_warning(f) else "  "
            line_count = f.line_end - f.line_start + 1
            sb.append(f"{f.name:<30} {f.cyclomatic_complexity:>8} "
                     f"{f.cognitive_complexity:>8} {f.nested_depth:>5} "
                     f"{line_count:>5} {warning}")

        sb.append(f"{'='*60}")

        # 生成建议
        warnings = [f for f in functions if self._is_warning(f)]
        if warnings:
            sb.append(f"\n🔴 发现 {len(warnings)} 个需要关注的函数：\n")
            for f in warnings[:5]:  # 最多显示5个
                reasons = []
                if f.cyclomatic_complexity > self.cyclomatic_threshold:
                    reasons.append(f"圈复杂度 {f.cyclomatic_complexity} 超过阈值 {self.cyclomatic_threshold}")
                if f.cognitive_complexity > self.cognitive_threshold:
                    reasons.append(f"认知复杂度 {f.cognitive_complexity} 超过阈值 {self.cognitive_threshold}")
                if (f.line_end - f.line_start) > self.max_lines:
                    reasons.append(f"函数行数 {f.line_end - f.line_start} 超过阈值 {self.max_lines}")

                sb.append(f"  • {f.name} (行 {f.line_start}-{f.line_end})")
                for r in reasons:
                    sb.append(f"    → {r}")

        return "\n".join(sb)

5.4 系统提示词设计

<!-- prompts/system.md -->
你是一个专业的代码审查员，拥有10年以上软件工程经验。

## 你的职责
1. 分析代码质量，发现潜在问题
2. 提供具体的改进建议
3. 评估代码的可维护性、可读性和性能
4. 识别安全漏洞和性能瓶颈

## 审查维度
- **正确性**：逻辑错误、边界条件处理、异常处理
- **可读性**：命名规范、注释质量、代码结构
- **可维护性**：耦合度、内聚性、复杂度
- **性能**：算法效率、内存使用、资源泄漏
- **安全**：输入验证、SQL注入、XSS、权限控制

## 输出格式
每次审查请按以下格式输出：

### 📋 概览
- 文件路径：xxx
- 总行数：xxx
- 警告项数：xxx

### ✅ 做得好的地方
列出代码中值得肯定的点（具体到行号）

### ⚠️ 需要改进的地方
按严重程度排序：
- 🔴 严重：必须修复
- 🟡 建议：推荐修复
- 🔵 参考：可选优化

### 📝 改进建议
针对每个问题，提供：
1. 问题描述
2. 具体位置（文件:行号）
3. 改进方案
4. 代码示例（before/after）

## 工作流程
1. 先用 read_file 查看代码（建议每次不超过100行）
2. 用 analyze_complexity 分析复杂度
3. 综合所有信息给出审查结论
4. 汇总成完整的审查报告

## 重要提醒
- 只评论代码，不要评论开发者的能力
- 提供可执行的改进建议，不要只说"需要重构"
- 复杂函数优先审查，优先处理高复杂度警告

5.5 运行效果

# 克隆目标仓库到工作目录
git clone https://github.com/your-org/target-project.git ./target_repo

# 启动 Agent
dsh run --task "请审查 target_repo/src 目录下的所有 Python 文件，重点关注复杂度较高的函数"

# Agent 执行过程（模拟输出）
#
# [Step 1] 读取代码结构
# → read_file: target_repo/src/models/user.py
# ← 成功，读取 245 行代码
#
# [Step 2] 复杂度分析
# → analyze_complexity: target_repo/src/models/user.py, show_warnings=true
# ← 发现 3 个警告项：
#    • create_user (行 45-78): 圈复杂度 14 ⚠️
#    • update_profile (行 102-156): 圈复杂度 18, 认知复杂度 22 ⚠️
#    • get_user_stats (行 180-198): 函数过长 48 行 ⚠️
#
# [Step 3] 深入分析高复杂度函数
# → read_file: target_repo/src/models/user.py, start_line=102, end_line=156, show_lines=true
# ← 发现 update_profile 函数有 7 层嵌套 if-else 分支
#
# ... (继续多步骤分析)
#
# [Step 15] 生成最终报告
# ✅ 代码审查完成！
#
# 📋 概览
# - 审查文件：4 个 .py 文件
# - 总行数：892 行
# - 发现警告：11 项（严重 2 项，建议 6 项，参考 3 项）
#
# 🔴 严重问题
# 1. target_repo/src/models/user.py:102 - update_profile 圈复杂度 18
#    → 建议：拆分为多个单一职责的子函数
# 2. target_repo/src/services/auth.py:45 - SQL 拼接存在注入风险
#    → 建议：使用参数化查询
#
# 🟡 建议项（6项）
# ...

六、生产环境调优：从玩具到工业级

6.1 性能瓶颈分析

DSH 在生产环境中常见的性能瓶颈有三个：

1. LLM 调用延迟

每次工具调用都要经过 LLM 推理 -> 工具执行 -> 结果解析 -> 上下文更新的循环。如果 LLM 延迟 2 秒，50 步任务就要 100 秒。

优化方案：

• 使用流式输出（Streaming），让 Agent 可以"边想边执行"
• 实现并行工具发现：一次 LLM 调用识别多个可以并行的工具
• 缓存常用工具的描述和 Schema

// engine/parallel_discovery.go
func (a *Agent) discoverParallelActions(prompt string) ([]ActionCandidate, error) {
    resp, err := a.model.DecideWithHints(prompt, DecisionHints{
        MaxActions: 3,           // 最多同时发现3个可并行操作
        ParallelAllowed: true,   // 明确告知可以并行
    })
    if err != nil {
        return nil, err
    }
    return resp.Candidates, nil
}

2. 工具执行 I/O

文件 I/O、网络请求是主要的阻塞点。DSH 的工具执行器支持超时控制和自动重试：

// engine/tool_executor.go
type ToolExecutor struct {
    tools    map[string]cordis.Tool
    timeout  time.Duration
    retries  int
}

func (te *ToolExecutor) Execute(ctx context.Context, name string, params map[string]interface{}) (*ToolResult, error) {
    tool := te.tools[name]
    var lastErr error

    for attempt := 0; attempt <= te.retries; attempt++ {
        if attempt > 0 {
            // 指数退避重试
            time.Sleep(time.Duration(math.Pow(2, float64(attempt))) * 100 * time.Millisecond)
        }

        execCtx, cancel := context.WithTimeout(ctx, te.timeout)
        defer cancel()

        result, err := tool.Execute(execCtx, params)
        if err == nil {
            return result, nil
        }
        lastErr = err

        // 检查错误类型：可重试 vs 不可重试
        if !isRetryable(err) {
            break
        }
    }

    return nil, fmt.Errorf("工具执行失败（已重试%d次）: %w", te.retries, lastErr)
}

3. 上下文膨胀

长任务中，每次工具调用结果都进入上下文，导致 token 消耗快速增长。DSH 的上下文压缩策略：

// context/compressor.go
type SemanticCompressor struct {
    embedder *EmbeddingModel
}

func (sc *SemanticCompressor) Compress(text string, targetTokens int) string {
    // 1. 语义分块：将文本按语义边界切分
    chunks := sc.semanticChunk(text)

    // 2. 提取关键信息：每个块提取核心信息点
    summaries := make([]string, len(chunks))
    for i, chunk := range chunks {
        summary := sc.extractKeyInfo(chunk)
        summaries[i] = summary
    }

    // 3. 组装压缩后的文本
    compressed := strings.Join(summaries, "\n---\n")
    if countTokens(compressed) <= targetTokens {
        return compressed
    }

    // 4. 如果仍超预算，递归压缩
    return sc.Compress(compressed, targetTokens)
}

func (sc *SemanticCompressor) extractKeyInfo(text string) string {
    prompt := fmt.Sprintf(`请用一句话总结以下文本的核心信息（保留关键数据）：

文本：
%s

要求：
- 不超过50字
- 保留关键数值和术语
- 不添加解释性内容`, text)

    resp := sc.llm.Generate(prompt)
    return resp.Text
}

6.2 安全性加固

Agent 系统的安全性比普通程序更重要——因为它有执行能力。以下是生产环境必须做的安全加固：

1. 插件权限分级

# harness.yaml
security:
  plugin_mode: "whitelist"    # 白名单模式：只允许显式声明的插件

  permission_levels:
    # Level 0: 只读信息获取
    read_only:
      - filesystem:read
      - web:fetch
      - web:search

    # Level 1: 受限写入
    restricted_write:
      - filesystem:write  (allowed_paths: ["./output", "./temp"])
      - terminal        (allowed_commands: ["git status", "python3"])

    # Level 2: 全权限（仅 CI 环境）
    full:
      - "*"

  default_level: "read_only"

2. 操作审计日志

// security/audit.go
type AuditLogger struct {
    writer  *os.File
    encoder *json.Encoder
}

type AuditEntry struct {
    Timestamp     time.Time `json:"timestamp"`
    SessionID     string    `json:"session_id"`
    AgentID       string    `json:"agent_id"`
    Action        string    `json:"action"`         // tool_call | context_update | plan_step
    ToolName      string    `json:"tool_name,omitempty"`
    Parameters    string    `json:"parameters,omitempty"`  // 脱敏后
    ResultStatus  string    `json:"result_status"`  // success | failure | timeout
    DurationMs    int64     `json:"duration_ms"`
    RiskLevel     string    `json:"risk_level"`     // low | medium | high | critical
}

func (al *AuditLogger) Log(entry *AuditEntry) {
    // 高风险操作实时告警
    if entry.RiskLevel == "critical" {
        al.sendAlert(entry)
    }
    al.encoder.Encode(entry)
}

3. 速率限制

// security/rate_limiter.go
type RateLimiter struct {
    toolCalls  map[string]*tokenbucket.Bucket
    mutex      sync.Mutex
}

func NewRateLimiter() *RateLimiter {
    rl := &RateLimiter{
        toolCalls: make(map[string]*tokenbucket.Bucket),
    }
    // 默认：每个工具每秒最多10次调用
    for _, tool := range AllTools() {
        rl.toolCalls[tool.Name()] = tokenbucket.New(10, 10)
    }
    return rl
}

func (rl *RateLimiter) Allow(toolName string) bool {
    rl.mutex.Lock()
    defer rl.mutex.Unlock()

    bucket, ok := rl.toolCalls[toolName]
    if !ok {
        return false
    }
    return bucket.Allow()
}

6.3 监控与可观测性

生产环境必须有完整的监控体系：

// observability/metrics.go
type MetricsCollector struct {
    stepLatencies    []float64
    toolCallCounts   map[string]int
    toolCallLatencies map[string][]float64
    errorRates       map[string]float64
    contextTokens    []int
}

func (m *MetricsCollector) RecordStep(step *AgentStep) {
    m.stepLatencies = append(m.stepLatencies, float64(step.DurationMs))
    m.contextTokens = append(m.contextTokens, step.ContextTokens)

    for tool, latency := range step.ToolLatencies {
        m.toolCallCounts[tool]++
        m.toolCallLatencies[tool] = append(m.toolCallLatencies[tool], latency)
    }
}

// 导出 Prometheus 指标
func (m *MetricsCollector) Export() *MetricsSnapshot {
    return &MetricsSnapshot{
        AvgStepLatency:   mean(m.stepLatencies),
        P99StepLatency:   percentile(m.stepLatencies, 0.99),
        TotalToolCalls:   sumCounts(m.toolCallCounts),
        ToolCallHeatmap:  m.toolCallCounts,
        AvgContextTokens: mean(m.contextTokens),
        ErrorRate:        calcErrorRate(m.errorRates),
    }
}

关键监控指标：

• 任务成功率：最终给出有用回答的任务比例
• 平均步数：完成任务需要的平均工具调用次数（反映规划效率）
• P99 延迟：最慢5%任务的响应时间
• Token 效率：每千 token 能完成多少有效工作

七、DSH vs 竞品：为什么这个框架值得关注

DSH 最核心的差异化在于两点：

1. Cordis 协议真正实现了"一切皆插件"：不是贴标签式的口号，而是有完整的接口定义、Schema 描述和生命周期管理。
2. 上下文管理层是认真设计的：三层记忆架构解决了长任务中信息丢失和噪音污染的问题，而不是粗暴地"截断前 N 条消息"。

八、总结与展望

DeepSeek Harness 的出现，标志着 Agent 框架从"能用"走向"好用"的转折点。它解决了几个长期困扰开发者的问题：

• 工具互操作性：Cordis 协议让不同来源的插件可以自由组合
• 长任务可靠性：三层记忆架构让 Agent 在复杂任务中不再"失忆"
• 生产可用性：安全隔离、审计日志、速率限制等企业级特性开箱即用

但 DSH 也有明显的不足：

• 文档不够完善：很多设计细节需要看源码才能理解
• 插件生态还小：目前内置插件覆盖的场景有限
• 调试工具缺失：出现问题时很难追踪 Agent 的决策链路

展望未来，DSH 的发展方向可能包括：

1. 插件市场：类似 npm 的插件注册与分发机制
2. 可视化调试器：类似 IDE 的断点、变量查看功能
3. 多 Agent 协作协议：定义 Agent 之间如何通信和协作
4. 本地模型优化：针对 DeepSeek 系列模型专项优化工具调用准确率

立即体验：

# 安装
pip install dsh-cli

# 创建你的第一个 Agent
dsh init my-first-agent
cd my-first-agent
dsh run --task "帮我分析当前目录下所有 Python 文件的复杂度"

# 查看官方文档
dsh docs

DeepSeek Harness 的野心不止于"又一个 Agent 框架"，它想做的是 AI Native 时代的基础设施标准。 Cordis 协议能否成为 Agent 世界的"Corda"——就像 USB 之于设备互联一样——还需要时间验证。但至少现在，它已经迈出了坚实的第一步。