DeepSeek Harness 深度拆解：当「一切皆插件」把 AI Agent 开发从框架锁定中解放出来——从 Cordis 微内核到生产级部署的全链路实战
背景：为什么需要一个 Harness？
2026年的AI编程智能体赛道，已经从「模型能力」的比拼，悄然转向「工程框架」的竞争。Claude有Claude Code，OpenAI有Agents SDK，Cursor有自研的Origin——每家都在围绕自己的模型构建一套绑定生态的Agent运行时。
但DeepSeek的思路截然不同。
2026年8月13日深夜，DeepSeek发布了DeepSeek Harness（社区简称DSH）v0.1开发者预览版，并同步以MIT协议开源。项目主页只有一行核心等式：
Model + Harness = Agent
模型负责思考和推理，Harness负责模型之外的一切工程活——工具调用、任务规划、执行调度、安全沙箱、文件操作、会话持久化。这套框架的核心理念叫「一切皆插件」：模型、工具、技能、会话、沙箱、文件系统、循环、编排和UI，全部是可插拔的插件，可以自由组合、自由替换。
这意味着什么？意味着你不需要被任何特定模型绑定——今天接DeepSeek，明天换Claude，后天接本地模型，改一行配置就够了。也意味着框架本身不提供Agent能力，它只提供组装Agent的骨架——真正的能力来自你插入的插件。
这和传统的SDK设计哲学完全不同。传统SDK是「我给你一套能力，你在我的规则里用」；Harness是「我给你一套规则，你把任何能力插进来」。这种设计思路，对于需要构建企业级AI Agent系统的工程师来说，是真正值得关注的方向。
本文将从架构原理、Cordis微内核实现、插件开发实战、安全模型分析、以及生产级部署五个维度，对DeepSeek Harness进行全链路深度拆解。
一、架构全景：从宏观设计看「一切皆插件」
1.1 传统Agent框架的结构性问题
在深入Harness之前，我们需要理解为什么现有的Agent框架存在结构性问题。
以LangChain为例，它的核心理念是「链」（Chain）——把LLM调用、工具执行、记忆管理串联成一条处理链。这种设计在简单场景下很好用，但随着项目规模增长，问题接踵而至：
紧耦合问题：LangChain的核心组件（LLMChain、Agent、Memory）相互依赖，当你想替换底层的LLM时，往往需要改动大量代码。或者说，当LangChain本身有bug时，你几乎无法绕过它。
能力边界模糊：什么算工具？什么算记忆？什么算输出解析？这些边界在LangChain里是模糊的，导致代码组织混乱，维护成本陡增。
测试困难：由于组件之间的隐式依赖，单独测试某个模块几乎不可能。
厂商锁定：一旦你基于某个框架构建了大量工具和prompt工程，迁移到其他框架的成本极高。
这些问题在2024-2025年的Agent开发浪潮中被反复暴露。工程师们开始意识到：与其在一个框架里修修补补，不如从架构层面重新设计。
1.2 Harness的设计哲学
DeepSeek Harness的回应是：把框架做薄，把能力做散。
┌─────────────────────────────────────────────────────────────┐
│ DeepSeek Harness │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Cordis 微内核 │ │
│ │ (插件加载 / 依赖注入 / 可逆副作用 / 生命周期管理) │ │
│ └──────────────────────────────────────────────────────┘ │
│ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │ Model │ │ Tool │ │Session │ │Sandbox │ │ Loop │ │
│ │Adapter │ │Registry│ │ Logger │ │ │ │ │ │
│ └────────┘ └────────┘ └────────┘ └────────┘ └────────┘ │
│ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │Skill │ │ File │ │Storage │ │ UI │ │Policy │ │
│ │Registry│ │System │ │ │ │ │ │ │ │
│ └────────┘ └────────┘ └────────┘ └────────┘ └────────┘ │
│ ...更多插件... │
└─────────────────────────────────────────────────────────────┘
Harness将Agent的所有能力拆解为独立包：
packages/llm/—— 模型适配器（支持多种LLM接入）packages/shell/—— 进程管理和命令执行packages/fs/—— 文件系统抽象packages/tools/—— 工具注册与管理packages/session/—— 会话日志（事件溯源）packages/sandbox/—— 安全沙箱packages/loop/—— Agent主循环packages/ui/—— 用户界面
每个包都是一个独立的插件，可以在配置文件中启用、禁用或替换，而无需修改框架核心代码。
1.3 核心等式的工程含义
「Model + Harness = Agent」这个等式背后有两层含义：
第一层：职责边界清晰。模型只负责推理——接收文本输入，输出文本（或结构化）响应。Harness负责所有工程任务——调用工具、管理记忆、处理IO、维护安全边界。这种分离使得测试变得极为简单：你只需要用一个mock LLM就能测试整个Harness的行为，而不需要一个真实的模型。
第二层：可替换性是架构层面的承诺。当DeepSeek说「一切皆插件」，他们不是在营销，而是真的在代码层面实现了这一点。每个插件都实现了标准接口，插件之间通过接口通信，而非直接依赖实现。这意味着：
- 你可以替换默认的模型适配器，改用自己的本地模型
- 你可以替换默认的工具注册表，接入自己的API生态
- 你可以替换默认的沙箱实现，使用更严格的安全策略
- 你甚至可以替换Agent Loop本身，用自己的编排逻辑
这种设计在工程上的收益是巨大的：团队可以在Harness之上构建差异化产品，而不需要fork和维护一个框架分支。
二、Cordis微内核：插件系统的地基
2.1 为什么需要一个专用的插件内核？
很多人听到「一切皆插件」的第一反应是：这不就是依赖注入吗？Spring IOC、Angular DI、Python的pluggy都能做。
但Harness的设计复杂度远超普通DI框架，因为它面临一个独特挑战：插件的卸载（unload）必须是完全可逆的。
在传统应用里，卸载一个模块通常意味着内存泄漏、资源未释放、状态不一致。但在Harness里，Agent是长时间运行的复杂系统，插件的动态加载和卸载是核心能力——你可能需要在运行时切换模型、替换工具集、甚至临时禁用某个危险操作。
这就引出了Cordis的核心创新：时空可组合性（Temporal-Spatial Composability）。
2.2 时间维度：可逆副作用
Cordis要求每个插件在执行任何副作用操作时，必须同时提供一个「逆操作」（reversal）。这不是简单的undo栈，而是一套形式化的机制：
# Cordis 插件示例：文件写入操作
class FileWritePlugin(Plugin):
def write(self, path: str, content: str) -> Effect:
"""
写入文件，返回一个包含正向操作和逆向操作的 Effect 对象
"""
# 记录操作前的状态（用于逆操作）
original = self.fs.read(path) if self.fs.exists(path) else None
# 执行正向操作
self.fs.write(path, content)
# 返回一个包含完整撤销信息的效果对象
return Effect(
forward=WriteOp(path=path, content=content),
reverse=WriteOp(path=path, content=original), # 逆操作
metadata={"timestamp": time.time()}
)
class Effect:
"""Cordis 的核心抽象：包含可逆副作用"""
forward: Operation # 正向操作
reverse: Operation # 逆向操作
metadata: dict # 元数据
def commit(self):
"""提交效果，永久化副作用"""
pass
def rollback(self):
"""回滚效果，执行逆操作"""
self.reverse.execute()
def defer(self, other: 'Effect'):
"""组合两个效果（先执行 self，再执行 other）"""
return ComposedEffect([self, other])
这套机制的意义在于：当一个插件被卸载时，Cordis可以按照正确的顺序执行所有相关Effect的逆操作，确保系统状态完全恢复到插件加载前的状态。不会出现文件残留、环境变量未清理、子进程未终止等常见问题。
2.3 空间维度：依赖关系管理
Cordis还需要管理插件之间的依赖关系。当多个插件都依赖同一个底层服务时，Cordis需要处理：
- 依赖冲突：两个插件需要不同版本的同一个依赖
- 循环依赖：插件A依赖B，B依赖C，C依赖A
- 初始化顺序：某些插件必须在其他插件之后才能启动
- 服务共享：多个插件共享同一个服务实例，避免重复创建
# Cordis 依赖声明与解析
class CordisKernel:
def __init__(self):
self.plugins: dict[str, Plugin] = {}
self.services: dict[str, Any] = {}
self.effect_log: list[Effect] = []
def load_plugin(self, plugin: Plugin) -> Result:
# 1. 检查依赖是否满足
for dep_name, dep_version in plugin.requires:
if dep_name not in self.services:
return Result.err(f"Missing dependency: {dep_name}")
if not self._check_version(self.services[dep_name], dep_version):
return Result.err(f"Version mismatch: {dep_name}")
# 2. 按拓扑排序确定初始化顺序
init_order = self._topological_sort(plugin.dependencies)
# 3. 执行初始化效果链
effects = []
for dep_plugin in init_order:
effect = dep_plugin.on_load(self)
effects.append(effect)
self.effect_log.append(effect)
# 4. 注册服务
self.services[plugin.name] = plugin.provides
self.plugins[plugin.name] = plugin
return Result.ok()
def unload_plugin(self, plugin_name: str) -> Result:
# 按逆序执行所有相关Effect的逆操作
plugin = self.plugins.pop(plugin_name)
# 找到所有与该插件相关的Effect
related_effects = [
e for e in reversed(self.effect_log)
if e.plugin_name == plugin_name
]
# 逆序回滚
for effect in related_effects:
effect.rollback()
del self.services[plugin.provides]
return Result.ok()
2.4 Cordis与Harness的关系
理解Cordis和Harness的关系很重要：Cordis是Harness的底层引擎，但它们是独立的两个项目。
- Cordis是一个通用的插件内核库，可以在任何Python项目中独立使用
- Harness建立在Cordis之上，专门针对AI Agent场景进行配置和封装
- Cordis处理「如何加载/卸载插件」和「如何管理副作用」
- Harness处理「这些插件如何组装成一个可用的Agent」
这种分层的好处是：Cordis可以独立演进，Harness可以基于相同的内核提供不同的配置和默认值。未来如果有人想构建一个非AI的插件化应用，也可以直接使用Cordis。
三、插件开发实战：从零构建一个自定义工具插件
3.1 环境准备与安装
先安装Harness。由于是早期预览版，推荐使用uv来管理依赖：
# 安装 uv（如果还没有）
curl -LsSf https://astral.sh/uv/install.sh | sh
# 创建项目并安装 DeepSeek Harness
uv init dsh-demo --python 3.12
cd dsh-demo
uv add deepseek-harness
# 安装完成后，查看项目结构
ls -la
# .venv/
# pyproject.toml
# hello.py
查看Harness的默认项目结构：
dsh init my-agent
cd my-agent
tree -L 2
# my-agent/
# ├── cordis.yml # 配置文件（核心！）
# ├── plugins/ # 本地插件目录
# ├── sessions/ # 会话日志存储
# └── README.md
cordis.yml是Harness的配置文件，它描述了Agent的所有行为：
# cordis.yml 示例配置
name: my-coding-agent
version: "1.0"
# 模型配置（这里可以替换成任何兼容的模型）
model:
adapter: deepseek-chat
api_key: ${DEEPSEEK_API_KEY}
base_url: https://api.deepseek.com
params:
temperature: 0.7
max_tokens: 4096
# 工具注册
tools:
enabled:
- shell # 命令行执行
- filesystem # 文件读写
- search # 搜索引擎
- web_fetch # 网页抓取
config:
shell:
timeout: 30
allowed_commands: ["git", "npm", "python", "cargo"]
filesystem:
root: ./workspace
max_file_size: 10485760 # 10MB
# 沙箱策略
sandbox:
mode: strict
allow_network: true
allow_filesystem_write: true
max_processes: 4
max_memory_mb: 2048
# 会话日志（事件溯源）
session:
backend: file
path: ./sessions
max_size_mb: 500
# Agent Loop 配置
loop:
max_iterations: 100
timeout_per_step: 60
confirmation_required:
- delete_file
- exec_command
- network_request
3.2 编写第一个自定义工具插件
假设我们要构建一个代码审查工具插件，用于自动分析代码质量并给出改进建议。
# plugins/code_review/__init__.py
"""
代码审查工具插件 - Code Review Plugin
演示如何在 DeepSeek Harness 中编写一个完整的自定义工具插件
"""
from cordis import Plugin, PluginContext, Effect, tool
class CodeReviewPlugin(Plugin):
"""
代码审查工具插件
提供的工具:
- analyze_code: 分析代码质量
- generate_review_report: 生成审查报告
- suggest_refactor: 提供重构建议
"""
name = "code_review"
version = "1.0.0"
provides = "code_review"
def __init__(self):
super().__init__()
self.issues: list[dict] = []
self.reports: list[dict] = []
def on_load(self, ctx: PluginContext) -> Effect:
"""插件加载时注册工具"""
return Effect(
forward=RegisterToolsEffect(
tools=[
tool(
name="analyze_code",
description="分析代码文件并识别质量问题",
parameters={
"file_path": {"type": "string", "required": True},
"language": {"type": "string", "required": False},
"strict_mode": {"type": "boolean", "required": False}
}
),
tool(
name="generate_review_report",
description="生成结构化代码审查报告",
parameters={
"issue_ids": {"type": "array", "required": False}
}
),
tool(
name="suggest_refactor",
description="针对指定问题提供重构建议",
parameters={
"issue_id": {"type": "string", "required": True}
}
)
]
),
reverse=ClearToolsEffect(tools=["analyze_code", "generate_review_report", "suggest_refactor"]),
plugin_name=self.name
)
@tool()
def analyze_code(self, ctx: PluginContext, file_path: str,
language: str = None, strict_mode: bool = False) -> dict:
"""
分析代码文件
Args:
file_path: 要分析的代码文件路径
language: 编程语言（auto-detect if None）
strict_mode: 严格模式（更多检查项）
Returns:
包含分析结果的字典
"""
# 读取文件
fs = ctx.get_service("filesystem")
if not fs.exists(file_path):
return {"error": f"File not found: {file_path}"}
content = fs.read(file_path)
if language is None:
language = self._detect_language(file_path)
# 执行分析
issues = self._perform_static_analysis(content, language, strict_mode)
issue_record = {
"id": self._generate_id(),
"file_path": file_path,
"language": language,
"issues": issues,
"severity": self._calculate_severity(issues),
"timestamp": ctx.now().isoformat()
}
self.issues.append(issue_record)
return {
"file": file_path,
"language": language,
"total_issues": len(issues),
"severity_breakdown": self._severity_summary(issues),
"issues": issues[:10], # 返回前10个问题，详细报告用 generate_review_report
"lines_of_code": len(content.splitlines())
}
def _perform_static_analysis(self, content: str, language: str, strict: bool) -> list[dict]:
"""执行静态代码分析"""
issues = []
lines = content.splitlines()
for i, line in enumerate(lines, 1):
# 检测常见问题
issues.extend(self._check_line(line, i, language, strict))
# 复杂度分析
complexity = self._analyze_cyclomatic_complexity(content, language)
if complexity > 10:
issues.append({
"type": "complexity",
"severity": "warning",
"line": 0,
"message": f"函数复杂度 {complexity} 超过阈值 10，建议拆分",
"suggestion": "考虑将函数拆分为更小的单元，每个函数只做一件事"
})
return issues
def _check_line(self, line: str, line_num: int, language: str, strict: bool) -> list[dict]:
issues = []
# 空行检测
if line.strip() == "":
if strict and line_num > 1:
issues.append({
"type": "style",
"severity": "info",
"line": line_num,
"message": "建议移除尾随空行" if not line.strip() else ""
})
# 行长度检查
if len(line) > 120:
issues.append({
"type": "style",
"severity": "info",
"line": line_num,
"message": f"行长度 {len(line)} 超过推荐长度 120",
"suggestion": "考虑拆分为多行或提取为变量"
})
# 语言特定检测
if language == "python":
# Tab vs 空格混用
if "\t" in line:
issues.append({
"type": "style",
"severity": "warning",
"line": line_num,
"message": "检测到 Tab 字符，建议使用空格",
"suggestion": "将 \\t 替换为 4 个空格"
})
# TODO 注释检测
if "# TODO" in line or "# FIXME" in line:
issues.append({
"type": "maintainability",
"severity": "info",
"line": line_num,
"message": "发现未完成的 TODO/FIXME",
"suggestion": "确保在提交前处理这些标记"
})
elif language == "javascript" or language == "typescript":
# console.log 检测
if "console.log" in line:
issues.append({
"type": "code_quality",
"severity": "warning",
"line": line_num,
"message": "生产代码中发现 console.log",
"suggestion": "使用结构化日志库（如 winston 或 pino）"
})
# any 类型检测（TypeScript）
if language == "typescript" and ": any" in line:
issues.append({
"type": "type_safety",
"severity": "error" if strict else "warning",
"line": line_num,
"message": "使用 'any' 类型绕过类型检查",
"suggestion": "使用 unknown 或具体类型替代"
})
return issues
def _analyze_cyclomatic_complexity(self, content: str, language: str) -> int:
"""简化版圈复杂度分析"""
complexity = 1 # 基础复杂度
patterns = [
r'\bif\b', r'\belse\b', r'\bfor\b', r'\bwhile\b',
r'\bcatch\b', r'\bcase\b', r'\?\s*:', r'&&', r'\|\|'
]
for pattern in patterns:
complexity += len(re.findall(pattern, content))
return complexity
def _detect_language(self, file_path: str) -> str:
ext_map = {
'.py': 'python', '.js': 'javascript', '.ts': 'typescript',
'.go': 'go', '.rs': 'rust', '.java': 'java',
'.cpp': 'cpp', '.c': 'c', '.rb': 'ruby', '.php': 'php'
}
import os
_, ext = os.path.splitext(file_path)
return ext_map.get(ext.lower(), 'unknown')
def _calculate_severity(self, issues: list[dict]) -> str:
if any(i.get('severity') == 'error' for i in issues):
return 'error'
if any(i.get('severity') == 'warning' for i in issues):
return 'warning'
return 'ok'
def _severity_summary(self, issues: list[dict]) -> dict:
summary = {"error": 0, "warning": 0, "info": 0}
for issue in issues:
sev = issue.get("severity", "info")
summary[sev] = summary.get(sev, 0) + 1
return summary
def _generate_id(self) -> str:
import uuid
return f"issue_{uuid.uuid4().hex[:8]}"
@tool()
def generate_review_report(self, ctx: PluginContext, issue_ids: list = None) -> dict:
"""生成结构化审查报告"""
if issue_ids:
issues = [i for i in self.issues if i["id"] in issue_ids]
else:
issues = self.issues[-1:] # 最近一次分析
if not issues:
return {"error": "No issues found. Run analyze_code first."}
report = {
"summary": {
"total_files": len(issues),
"total_issues": sum(len(i["issues"]) for i in issues),
"severity_overall": max((i["severity"] for i in issues), default="ok")
},
"files": []
}
for issue in issues:
file_report = {
"file": issue["file_path"],
"language": issue["language"],
"issues": issue["issues"],
"summary_by_severity": self._severity_summary(issue["issues"]),
"timestamp": issue["timestamp"]
}
report["files"].append(file_report)
self.reports.append(report)
return report
@tool()
def suggest_refactor(self, ctx: PluginContext, issue_id: str) -> dict:
"""针对指定问题提供重构建议"""
issue = next((i for i in self.issues for seg in [i.get("issues", [])]
if any(seg_item.get("id") == issue_id for seg_item in seg)), None)
if not issue:
return {"error": f"Issue {issue_id} not found"}
# 调用模型生成智能重构建议
model = ctx.get_service("model")
prompt = f"""针对以下代码问题，提供具体的重构建议：
问题类型: {issue['type']}
严重程度: {issue['severity']}
代码位置: {issue['file_path']}:{issue.get('line', 'N/A')}
问题描述: {issue['message']}
建议: {issue.get('suggestion', '无')}
请提供:
1. 问题根本原因分析
2. 具体代码级别的重构方案
3. 重构前后的代码对比示例
"""
response = model.complete(prompt)
return {"issue_id": issue_id, "suggestion": response}
# 插件入口点
plugin = CodeReviewPlugin
3.3 在 cordis.yml 中注册插件
将自定义插件集成到Harness配置中：
# cordis.yml 中添加自定义插件
name: my-coding-agent
version: "1.0"
# 启用内置工具
tools:
enabled:
- shell
- filesystem
- code_review # 自定义插件
# 插件配置
plugins:
code_review:
path: ./plugins/code_review
config:
strict_mode: true
max_issues_per_file: 50
excluded_patterns:
- "**/test_*.py"
- "**/__pycache__/**"
- "**/node_modules/**"
3.4 与Agent Loop的集成
Harness的Agent Loop会自动发现并注册所有工具。Agent执行时，会看到类似这样的工具列表：
可用工具:
1. shell.execute(command: string) -> string
2. filesystem.read(path: string) -> string
3. filesystem.write(path: string, content: string) -> void
4. code_review.analyze_code(file_path: string, language?: string, strict_mode?: boolean) -> dict
5. code_review.generate_review_report(issue_ids?: string[]) -> dict
6. code_review.suggest_refactor(issue_id: string) -> dict
Agent可以在一次对话中自由调用这些工具，就像使用一个完整的开发环境一样：
用户: 帮我审查一下 src/utils/parser.py 这个文件
Agent:
→ code_review.analyze_code(file_path="src/utils/parser.py", strict_mode=true)
→ 返回: 发现3个warning，5个info，代码复杂度为8
→ code_review.generate_review_report()
→ 生成完整报告并展示给用户
四、事件溯源与会话日志：为什么「Model-Visible Means Logged」
4.1 Agent的可观测性问题
传统的Agent系统在调试时面临巨大挑战：当Agent给出错误的回答时，你很难知道它是怎么得出这个结论的。是因为prompt不够清晰？工具调用参数错误？模型推理出现了幻觉？还是某次API调用返回了意外的数据？
DeepSeek Harness采用了事件溯源（Event Sourcing）模式来解决这个问题：所有对模型可见的信息，都必须被记录到会话日志中。这就是Harness的设计原则：Model-Visible Means Logged。
4.2 会话日志的数据结构
# 会话日志条目示例
@dataclass
class SessionEvent:
event_id: str # 唯一标识符（雪花ID）
timestamp: datetime # 事件发生时间
session_id: str # 所属会话
event_type: str # 事件类型
data: dict # 事件数据
# 事件类型枚举
# USER_MESSAGE - 用户输入
# MODEL_OUTPUT - 模型输出
# TOOL_CALL - 工具调用请求
# TOOL_RESULT - 工具执行结果
# ERROR - 错误信息
# SYSTEM_MESSAGE - 系统消息
class SessionEvent:
def __init__(
self,
session_id: str,
event_type: str,
data: dict,
parent_event_id: str = None
):
self.event_id = snowflake_id()
self.timestamp = datetime.now(timezone.utc)
self.session_id = session_id
self.event_type = event_type
self.data = data
self.parent_event_id = parent_event_id
def to_json(self) -> str:
return json.dumps({
"event_id": self.event_id,
"timestamp": self.timestamp.isoformat(),
"session_id": self.session_id,
"event_type": self.event_type,
"data": self.data,
"parent_event_id": self.parent_event_id
}, ensure_ascii=False)
# 示例：一个完整的Agent执行轨迹
events = [
SessionEvent(
session_id="sess_abc123",
event_type="USER_MESSAGE",
data={
"content": "帮我审查一下 src/auth/login.py"
}
),
SessionEvent(
session_id="sess_abc123",
event_type="MODEL_OUTPUT",
data={
"content": "好的，我来审查这个文件。让我先调用分析工具...",
"tool_calls": [{
"id": "call_001",
"name": "code_review.analyze_code",
"arguments": {"file_path": "src/auth/login.py", "strict_mode": True}
}]
},
parent_event_id="call_001" # 模型调用了工具
),
SessionEvent(
session_id="sess_abc123",
event_type="TOOL_RESULT",
data={
"tool_call_id": "call_001",
"result": {
"file": "src/auth/login.py",
"language": "python",
"total_issues": 4,
"severity_breakdown": {"error": 1, "warning": 2, "info": 1}
}
},
parent_event_id="call_001" # 工具返回了结果
),
SessionEvent(
session_id="sess_abc123",
event_type="MODEL_OUTPUT",
data={
"content": "审查完成！发现了一个严重问题...\n\n**问题1 [Error]**：SQL注入风险\n文件: src/auth/login.py:23\n```python\nquery = f\"SELECT * FROM users WHERE username='{username}'\"\n```\n这条SQL直接拼接了用户输入，存在SQL注入风险。建议使用参数化查询..."
}
)
]
4.3 会话日志的append-only特性
Harness的会话日志是严格append-only的。这意味着：
- 日志一旦写入，就永远不会被修改或删除
- 历史状态通过重放（replay）事件来重建
- 这使得完整的执行轨迹可以被重现、审查和回放
class AppendOnlySessionLog:
"""
只能追加、不能修改的会话日志
基于文件或数据库实现持久化存储
"""
def __init__(self, storage_path: str):
self.storage_path = storage_path
self._ensure_directory()
def append(self, event: SessionEvent) -> str:
"""追加事件（唯一允许的写操作）"""
# 追加到文件
filepath = os.path.join(
self.storage_path,
f"{event.session_id}.jsonl" # JSONL格式，每行一个事件
)
with open(filepath, "a", encoding="utf-8") as f:
f.write(event.to_json() + "\n")
return event.event_id
def replay(self, session_id: str) -> list[SessionEvent]:
"""重放整个会话，重建所有历史状态"""
filepath = os.path.join(self.storage_path, f"{session_id}.jsonl")
if not os.path.exists(filepath):
return []
events = []
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
data = json.loads(line)
events.append(SessionEvent(**data))
return events
def rebuild_state(self, session_id: str) -> SessionState:
"""从事件历史重建当前状态"""
events = self.replay(session_id)
state = SessionState()
for event in events:
state.apply(event) # 逐步应用每个事件
return state
def search(self, session_id: str,
event_types: list[str] = None,
keyword: str = None) -> list[SessionEvent]:
"""查询会话事件（支持类型过滤和关键词搜索）"""
events = self.replay(session_id)
if event_types:
events = [e for e in events if e.event_type in event_types]
if keyword:
events = [
e for e in events
if keyword.lower() in json.dumps(e.data).lower()
]
return events
4.4 append-only设计的工程收益
这种设计带来了几个重要收益：
1. 完整的可观测性：任何时刻都可以回溯Agent的完整思考链。模型收到了什么？调用了什么工具？工具返回了什么？每一步都清清楚楚。
2. 精确的错误定位：当Agent出错时，可以精确地定位到是哪一步出了问题——是模型推理错误？是工具返回了脏数据？还是工具调用本身失败了？
3. 合规与审计：在企业场景中，append-only日志满足了数据合规要求——所有操作都有完整记录，不可篡改。
4. Agent自进化：基于完整的执行日志，可以训练更强大的Agent——分析成功案例学习好的策略，分析失败案例避免坏的决策。
5. 会话恢复：如果Agent进程崩溃，可以从日志中恢复会话状态，继续之前的工作。
五、四种运行模式：如何在不同场景下使用Harness
DeepSeek Harness提供了四种开箱即用的运行模式，适用于不同的使用场景：
5.1 极简模式（Minimal Mode）
适合快速尝鲜和本地开发。极简模式只需要一个cordis.yml和一个API Key：
# 5行配置启动一个编程Agent
cat > cordis.yml << 'EOF'
name: quick-agent
model:
adapter: deepseek-chat
api_key: ${DEEPSEEK_API_KEY}
tools:
enabled: [shell, filesystem]
EOF
dsh run
# → 启动交互式Agent会话
5.2 终端模式（Terminal Mode）
适合需要深度集成的开发者。Terminal模式提供了完整的TUI界面：
┌─────────────────────────────────────────────────────────┐
│ DeepSeek Harness - Terminal Mode │
│ Session: sess_abc123 | Model: deepseek-chat │
├─────────────────────────────────────────────────────────┤
│ [Tool: shell] $ git status │
│ On branch main │
│ Your branch is up to date with 'origin/main'. │
│ │
│ [Tool: filesystem] Read: src/main.py (45 lines) │
│ ✓ Code review: 2 warnings, 1 info │
│ │
│ [Model] 分析完成。主要问题在第23行... │
├─────────────────────────────────────────────────────────┤
│ > _ │
└─────────────────────────────────────────────────────────┘
5.3 服务模式（Service Mode）
适合构建API服务，将Harness暴露为REST/WebSocket API：
# service.py - 将Harness作为服务运行
from dsh import Harness
harness = Harness.from_config("cordis.yml")
@app.post("/agent/execute")
async def execute(request: ExecuteRequest):
"""通过API执行Agent任务"""
result = await harness.execute(
prompt=request.prompt,
context=request.context,
session_id=request.session_id
)
return {"result": result}
@app.get("/agent/sessions/{session_id}/events")
async def get_events(session_id: str, event_type: str = None):
"""获取会话事件日志"""
events = harness.session_log.search(
session_id=session_id,
event_types=[event_type] if event_type else None
)
return {"events": [e.data for e in events]}
5.4 浏览器模式（Browser Mode）
适合需要图形界面的场景。Harness内置了一个轻量级的Web UI：
# 启动浏览器模式
dsh run --mode browser --port 8080
# → 打开 http://localhost:8080 访问Web界面
浏览器模式支持：
- 可视化的工具调用流程
- 会话历史和事件回放
- 实时流式输出（Streaming）
- 文件编辑器集成
六、安全模型：插件、沙箱与信任边界
6.1 安全问题的重要性
由于Harness运行在用户的机器上，可以执行任意命令、读写文件系统、访问网络，它的安全模型至关重要。如果一个恶意的工具插件可以劫持Agent的行为，后果不堪设想。
Harness的安全模型包含三个层面：沙箱隔离、权限策略、审计追踪。
6.2 沙箱隔离
Harness使用多种沙箱技术来隔离危险操作：
进程级隔离：工具执行在独立的子进程中运行，有资源限制：
sandbox:
mode: strict
max_processes: 4 # 最多4个并发进程
max_memory_mb: 2048 # 每个进程最多2GB内存
max_cpu_percent: 80 # 最多占用80% CPU
max_execution_time: 300 # 单次执行最多5分钟
network_policy: allowlist # 白名单网络策略
allowed_domains:
- "api.github.com"
- "api.deepseek.com"
文件系统隔离：工具只能访问配置中指定的工作目录：
# 文件系统沙箱实现
class FilesystemSandbox:
def __init__(self, root_dir: str, allowed_paths: list[str] = None):
self.root_dir = os.path.abspath(root_dir)
self.allowed_paths = [os.path.abspath(p) for p in (allowed_paths or [])]
def resolve_path(self, path: str) -> str:
"""解析并验证路径是否在沙箱内"""
abs_path = os.path.abspath(os.path.join(self.root_dir, path))
# 检查是否在允许路径内
for allowed in self.allowed_paths:
if abs_path.startswith(allowed):
return abs_path
# 检查是否在root_dir内
if not abs_path.startswith(self.root_dir):
raise SecurityError(f"Path escape attempt: {path}")
return abs_path
def read(self, path: str) -> str:
resolved = self.resolve_path(path)
if not os.path.exists(resolved):
raise FileNotFoundError(path)
return open(resolved, "r", encoding="utf-8").read()
def write(self, path: str, content: str):
resolved = self.resolve_path(path)
# 禁止覆盖系统文件
if self._is_protected_path(resolved):
raise SecurityError(f"Protected path: {path}")
open(resolved, "w", encoding="utf-8").write(content)
def _is_protected_path(self, path: str) -> bool:
protected = ["/etc/passwd", "/etc/shadow", "~/.ssh/", "~/.aws/"]
return any(p in path for p in protected)
网络隔离：工具只能访问白名单中的域名和端口：
class NetworkSandbox:
def __init__(self, allowed_domains: list[str]):
self.allowed_domains = set(allowed_domains)
self._block_other_outgoing()
def _block_other_outgoing(self):
"""通过iptables/nftables阻止非白名单流量"""
import subprocess
# 仅示例：实际实现需要根据平台选择不同方案
subprocess.run([
"iptables", "-A", "OUTPUT", "-j", "DROP"
], check=False)
def check_request(self, url: str) -> bool:
from urllib.parse import urlparse
parsed = urlparse(url)
domain = parsed.netloc.split(":")[0]
return domain in self.allowed_domains
6.3 权限策略：确认机制
对于高风险操作，Harness实现了交互式确认机制：
class ConfirmationPolicy:
"""高风险操作的确认策略"""
ALWAYS_CONFIRM = ["delete_file", "exec_command", "network_request", "env_write"]
CONFIRM_THRESHOLD = {"shell": {"risk_score": 7}, "filesystem": {"risk_score": 8}}
def __init__(self, auto_approved: bool = False):
self.auto_approved = auto_approved
def requires_confirmation(self, tool_name: str, arguments: dict) -> bool:
if self.auto_approved:
return False
# 检查是否是高风险工具
if tool_name in self.ALWAYS_CONFIRM:
return True
# 评估风险分数
risk_score = self._calculate_risk_score(tool_name, arguments)
threshold = self.CONFIRM_THRESHOLD.get(tool_name, {}).get("risk_score", 5)
return risk_score >= threshold
def _calculate_risk_score(self, tool_name: str, arguments: dict) -> int:
score = 0
if tool_name == "shell":
cmd = arguments.get("command", "")
# 危险命令检测
if any(dangerous in cmd for dangerous in ["rm -rf", "chmod 777", "shutdown", "reboot"]):
score += 10
if "sudo" in cmd:
score += 5
elif tool_name == "filesystem":
if arguments.get("operation") == "delete":
score += 8
if arguments.get("path", "").startswith("/etc"):
score += 10
return score
# 使用示例
policy = ConfirmationPolicy(auto_approved=False)
async def execute_tool_call(tool_call: ToolCall, ctx: PluginContext):
if policy.requires_confirmation(tool_call.name, tool_call.arguments):
# 需要用户确认
confirmed = await ctx.request_confirmation(
title=f"确认执行: {tool_call.name}",
details=f"参数: {tool_call.arguments}",
risk_level="high" if policy._calculate_risk_score(
tool_call.name, tool_call.arguments
) >= 7 else "medium"
)
if not confirmed:
return {"error": "Operation cancelled by user"}
# 执行工具...
6.4 已披露的安全漏洞分析
值得注意的是，2026年8月14日，有安全研究员披露了Harness v0.1预览版的四个安全漏洞（来源：企鹅号技术分析文章）：
- 插件注入攻击：恶意配置可以通过插件注册机制注入未授权的代码执行路径
- 沙箱逃逸：某些工具组合可以绕过文件系统沙箱限制
- 主密钥滥用：某些调试/管理接口可以被滥用以获取系统权限
- 事件日志重放攻击：append-only日志在特定条件下可以被伪造
这些漏洞的存在是预览版的正常现象。DeepSeek在官方文档中也明确标注了这些已知的限制，并承诺在正式版中修复。
对于开发者来说，在正式版发布前，应该：
- 避免在生产环境直接暴露Harness的API
- 使用网络隔离和进程隔离保护运行主机
- 对插件来源进行充分的安全审查
- 关注官方的安全更新公告
七、生产级部署：从开发到上线的完整指南
7.1 环境变量与密钥管理
在生产环境中，绝对不要将API密钥硬编码在配置文件中：
# 使用环境变量
export DEEPSEEK_API_KEY="sk-xxxx"
export CORDIS_LOG_LEVEL="info"
export CORDIS_SESSION_PATH="/data/sessions"
# 或者使用密钥管理服务
export CORDIS_KMS_BACKEND="aws_secrets_manager"
export AWS_SECRETS_MANAGER_SECRET_ID="dsh/api-keys"
推荐使用专门的密钥管理工具：
- 本地开发：1Password CLI、
pass或.env文件（需要加入.gitignore） - 服务器部署：AWS Secrets Manager、HashiCorp Vault、Azure Key Vault
- 容器环境：Kubernetes Secrets配合External Secrets Operator
7.2 Docker化部署
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
git curl iptables \
&& rm -rf /var/lib/apt/lists/*
# 安装 uv 和 Harness
RUN pip install uv
COPY pyproject.toml .
RUN uv sync --frozen
# 复制配置（注意：不要包含密钥）
COPY cordis.yml .
COPY plugins/ ./plugins/
# 创建非root用户
RUN useradd -m -u 1000 agent
USER agent
# 会话存储目录
RUN mkdir -p /data/sessions && chown agent:agent /data/sessions
ENV CORDIS_SESSION_PATH=/data/sessions
CMD ["dsh", "run", "--mode", "service"]
# docker-compose.yml
services:
harness:
build: .
ports:
- "8080:8080"
environment:
- DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY}
- CORDIS_LOG_LEVEL=info
volumes:
- ./sessions:/data/sessions
- /var/run/docker.sock:/var/run/docker.sock # 如果需要Docker in Docker
ulimits:
nofile:
soft: 65536
hard: 65536
deploy:
resources:
limits:
cpus: '2'
memory: 4G
reservations:
cpus: '0.5'
memory: 1G
restart: unless-stopped
# 配合 nginx 做反向代理和限流
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
depends_on:
- harness
7.3 监控与可观测性
在生产环境中，需要对Harness进行全面的监控：
# metrics_plugin.py - Prometheus 指标采集
from dsh import Plugin, PluginContext, Effect
import prometheus_client as prom
class MetricsPlugin(Plugin):
"""Prometheus 指标采集插件"""
name = "metrics"
provides = "metrics"
def __init__(self):
self.counter_tool_calls = prom.Counter(
"harness_tool_calls_total",
"Total tool calls",
["tool_name", "status"]
)
self.histogram_latency = prom.Histogram(
"harness_tool_latency_seconds",
"Tool call latency",
["tool_name"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60]
)
self.gauge_sessions = prom.Gauge(
"harness_active_sessions",
"Number of active sessions"
)
self.counter_errors = prom.Counter(
"harness_errors_total",
"Total errors",
["error_type"]
)
def on_load(self, ctx: PluginContext) -> Effect:
# 注册指标端点
ctx.on("tool_call_start", self._on_tool_call_start)
ctx.on("tool_call_end", self._on_tool_call_end)
ctx.on("error", self._on_error)
return Effect(forward=NoOp(), reverse=NoOp(), plugin_name=self.name)
@prom.Timer()
def _on_tool_call_start(self, event):
self.counter_tool_calls.labels(
tool_name=event.data["tool_name"],
status="started"
).inc()
def _on_tool_call_end(self, event):
status = "success" if "error" not in event.data else "error"
self.counter_tool_calls.labels(
tool_name=event.data["tool_name"],
status=status
).inc()
def _on_error(self, event):
self.counter_errors.labels(
error_type=event.data.get("type", "unknown")
).inc()
7.4 高可用架构
对于需要高可用的生产环境，推荐以下架构：
┌─────────────┐
│ Nginx │
│ (负载均衡) │
└──────┬──────┘
│
┌────────────────┼────────────────┐
│ │ │
┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐
│ Harness │ │ Harness │ │ Harness │
│ Instance 1│ │ Instance 2│ │ Instance 3│
└─────┬─────┘ └─────┬─────┘ └─────┬─────┘
│ │ │
└────────────────┼────────────────┘
│
┌───────────┴───────────┐
│ │
┌─────▼─────┐ ┌─────▼─────┐
│ Redis │ │ PostgreSQL│
│ (会话共享) │ │ (事件存储) │
└───────────┘ └───────────┘
- 水平扩展：多个Harness实例通过Nginx负载均衡
- 会话共享：使用Redis存储活跃会话状态，支持会话在实例间迁移
- 事件持久化：PostgreSQL存储会话事件，支持长期查询和分析
- 自动扩缩容：根据负载指标自动增减实例数量
八、与竞品对比：Harness的独特价值
8.1 技术定位对比
| 维度 | DeepSeek Harness | LangChain | LangGraph | CrewAI |
|---|---|---|---|---|
| 架构哲学 | 一切皆插件 | 链式组合 | 图结构编排 | Agent团队 |
| 模型绑定 | 无（可替换） | 松耦合 | 松耦合 | OpenAI优先 |
| 沙箱安全 | 内置多层沙箱 | 无 | 无 | 无 |
| 事件溯源 | append-only日志 | 无 | 无 | 有限 |
| 插件系统 | Cordis微内核 | 第三方 | 第三方 | 内部实现 |
| 开源协议 | MIT | MIT | MIT | Apache 2.0 |
| 成熟度 | 预览版(0.1) | 生产级 | 生产级 | 生产级 |
8.2 核心差异分析
Harness vs LangChain：LangChain的核心理念是「组合」，但它的组件之间有大量隐式依赖。Harness的核心理念是「替换」——每个组件都可以被替换而不影响其他组件。这对于需要构建差异化产品的团队来说，是更灵活的设计。
Harness vs LangGraph：LangGraph用图结构来表达Agent的工作流，适合复杂的多步骤推理场景。Harness则更关注Agent的「运行时」——如何让Agent在真实环境中可靠地执行任务。两者可以互补：LangGraph定义工作流，Harness提供执行底座。
Harness vs CrewAI：CrewAI专注于多Agent协作场景，Harness则是一个更通用的框架。CrewAI的「Agent团队」概念在Harness中可以通过插件组合来实现，但需要更多的配置工作。
九、局限性与发展展望
9.1 当前版本的局限
作为一个v0.1预览版，Harness存在以下明显局限：
- 稳定性不足：官方明确警告「可能有破坏性变更」，不适合直接用于生产环境
- 文档不完整：很多高级功能的API尚未完整文档化
- 插件生态薄弱：目前官方提供的插件数量有限，社区生态尚未形成
- 性能未经优化：早期版本在长会话场景下可能有内存和性能问题
- Windows支持有限：部分沙箱功能在Windows上实现不完整
9.2 未来发展方向
从已有信息推断，Harness的未来发展可能包括：
- 插件市场：类似VS Code插件市场的插件分发平台
- 可视化编排：拖拽式的Agent工作流设计器
- 多模态支持：图像、音频、视频等多模态工具
- 分布式Agent：跨多台机器协作的Agent网络
- 正式版LTS：稳定的生产级版本承诺
总结：Harness给AI Agent开发带来了什么？
DeepSeek Harness的核心价值，可以归纳为三点：
1. 解除厂商锁定：通过「一切皆插件」的架构，开发者第一次可以在不修改应用代码的情况下，任意切换底层模型和工具链。这对于需要保持技术灵活性的团队来说，是真正的架构级自由。
2. 引入企业级工程实践：事件溯源、可逆副作用、多层沙箱——这些原本只在金融和分布式系统领域使用的工程实践，被引入到AI Agent开发中。这意味着Agent系统第一次可以像其他关键业务系统一样，被认真地工程化。
3. 降低Agent开发门槛：通过标准化的插件接口和开箱即用的运行模式，Harness让「组装一个能真正干活的Agent」变得前所未有的简单。你不需要理解复杂的LangChain概念，不需要编写冗长的配置，只需要组合插件。
当然，v0.1预览版的状态意味着它还不适合直接用于生产。但它的架构设计已经展示了一种可能：未来的AI Agent开发，可能不再需要选择框架，而是选择插件。就像今天的Web开发不再绑定特定的服务器软件，而是通过标准接口组合各种中间件一样。
DeepSeek Harness迈出了这一步。值得密切关注。
Tags: DeepSeek|Harness|Cordis|AI Agent|插件化架构|事件溯源|沙箱安全|生产部署|Python|开源框架
Keywords: DeepSeek Harness|Cordis微内核|一切皆插件|AI Agent框架|事件溯源|可逆副作用|插件系统|安全沙箱|生产级部署