OpenHarness 深度研究 | 模块二：技术架构深度解析

2.1 整体架构概览：10个子系统的协同设计

OpenHarness 的核心架构设计体现了”模块化解耦、职责单一”的工程原则。整个系统被划分为 10 个逻辑子系统，每个子系统负责特定的功能领域，通过清晰的接口进行协作。

2.1.1 子系统职责矩阵

flowchart TB
    subgraph "OpenHarness Architecture"
        UI["🖥️ UI (React/Ink TUI)"] --> Engine["🧠 Engine<br/>Agent Loop"]
        Engine --> Tools["🔧 Tools<br/>43+ Tools"]
        Engine --> Skills["📚 Skills<br/>Knowledge Loading"]
        Engine --> Plugins["🔌 Plugins<br/>Extensions"]
        Engine --> Permissions["🛡️ Permissions<br/>Safety"]
        Engine --> Hooks["⚡ Hooks<br/>Lifecycle"]
        Engine --> Commands["💬 Commands<br/>54 Commands"]
        Engine --> MCP["🌐 MCP<br/>Protocol Client"]
        Engine --> Memory["🧠 Memory<br/>Persistence"]
        Engine --> Tasks["📋 Tasks<br/>Background Jobs"]
        Engine --> Coordinator["🤝 Coordinator<br/>Multi-Agent"]
    end
    
    Tools --> Config["⚙️ Config<br/>Settings"]
    Skills --> Config
    Plugins --> Config
    
    style Engine fill:#4a90d9,stroke:#333,stroke-width:2px,color:#fff
    style UI fill:#61dafb,stroke:#333,stroke-width:1px
    style Tools fill:#f0ad4e,stroke:#333,stroke-width:1px
    style Permissions fill:#d9534f,stroke:#333,stroke-width:1px

Caption: OpenHarness 10个子系统架构图。Engine 作为核心协调器，连接其他9个子系统，形成完整的 Agent Harness 能力。

2.1.2 子系统详细说明

代码规模基于 GitHub 仓库目录结构估算

2.1.3 设计原则：为什么这样划分？

OpenHarness 的子系统划分遵循了以下设计原则：

1. 单一职责原则（SRP）

每个子系统只负责一个明确的功能领域。例如，permissions 子系统只关心”是否允许执行”，而不关心”如何执行”。这种划分使得：

• 代码更容易理解和测试
• 修改一个子系统不会影响其他子系统
• 新开发者可以快速定位相关代码

2. 依赖倒置原则（DIP）

Engine 作为核心，依赖其他子系统的抽象接口，而不是具体实现。这使得：

• 工具、技能、插件可以独立演进
• 用户可以替换默认实现（如自定义权限策略）
• 单元测试更容易进行 Mock

3. 开闭原则（OCP）

系统对扩展开放，对修改关闭。例如：

• 新增工具只需继承 BaseTool，无需修改 engine
• 新增技能只需添加 .md 文件，无需修改代码
• 新增插件遵循标准接口，系统自动识别

2.2 Agent Loop：核心执行引擎

2.2.1 The Loop 的本质

Agent Loop 是 OpenHarness 的心脏，它实现了经典的 “Observe → Think → Act” 循环。与简单的请求-响应模式不同，Agent Loop 允许 LLM 进行多轮工具调用，直到任务完成。

sequenceDiagram
    participant User
    participant Engine
    participant LLM
    participant Tools
    participant Permissions
    
    User->>Engine: 提交请求
    loop Agent Loop
        Engine->>LLM: stream(messages, tools)
        LLM-->>Engine: tool_use / content
        
        alt stop_reason == "tool_use"
            Engine->>Permissions: 检查权限
            Permissions-->>Engine: 允许/拒绝
            
            Engine->>Tools: execute(tool_call)
            Tools-->>Engine: ToolResult
            
            Engine->>Engine: 更新 messages
        else stop_reason != "tool_use"
            Engine-->>User: 返回最终结果
        end
    end

Caption: Agent Loop 执行流程。循环持续进行直到 LLM 不再请求工具调用，表示任务完成。

2.2.2 核心代码实现分析

OpenHarness 的 Agent Loop 实现体现了 Python 异步编程的最佳实践：

# 简化版核心逻辑（基于 OpenHarness 架构）
async def agent_loop(
    self,
    messages: List[Message],
    tools: List[BaseTool],
    max_turns: int = 100
) -> AgentResponse:
    """
    核心的 Agent Loop 实现
    
    循环直到：
    1. LLM 不再请求工具调用（stop_reason != "tool_use"）
    2. 达到最大轮次限制
    3. 用户中断
    """
    turn_count = 0
    
    while turn_count < max_turns:
        # 1. 流式调用 LLM
        response = await self.llm.stream(
            messages=messages,
            tools=[t.to_json_schema() for t in tools]
        )
        
        # 2. 检查是否需要工具调用
        if response.stop_reason != "tool_use":
            # 任务完成，返回结果
            return AgentResponse(
                content=response.content,
                tool_calls=[]
            )
        
        # 3. 执行工具调用
        tool_results = []
        for tool_call in response.tool_uses:
            # 3.1 权限检查
            permission = await self.permissions.check(tool_call)
            if not permission.allowed:
                result = ToolResult(
                    error=f"Permission denied: {permission.reason}"
                )
            else:
                # 3.2 执行前 Hook
                await self.hooks.run_pre_tool_use(tool_call)
                
                # 3.3 执行工具
                result = await self.tools.execute(
                    tool_call,
                    context=self.context
                )
                
                # 3.4 执行后 Hook
                await self.hooks.run_post_tool_use(tool_call, result)
            
            tool_results.append(result)
        
        # 4. 更新消息历史
        messages.extend(self._format_tool_results(tool_results))
        turn_count += 1
    
    # 达到最大轮次限制
    return AgentResponse(
        content="Maximum turns reached",
        tool_calls=[],
        truncated=True
    )

关键设计亮点：

1. 流式处理（Streaming）：通过 stream() API 实现实时响应，用户体验更佳
2. 并发工具执行：多个独立工具可以并行执行，提高效率
3. 权限检查前置：每个工具调用前都进行权限验证，确保安全
4. Hook 机制：Pre/Post 钩子允许插件介入执行流程
5. 状态管理：messages 列表维护完整的对话上下文

2.2.3 Harness 方程解读

OpenHarness 提出了一个简洁的 Harness 定义方程：

方程各项的含义：

这个方程的价值在于：它明确了 Agent Harness 的核心组成要素，为框架设计和评估提供了理论框架。

2.3 Tools 子系统：43+工具的实现机制

2.3.1 工具分类与功能映射

OpenHarness 实现了 43+ 个工具，分为9个功能类别：

mindmap
  root((Tools<br/>43+))
    FileIO["🔧 File I/O<br/>6 tools"]
      Bash
      Read
      Write
      Edit
      Glob
      Grep
    Search["🔍 Search<br/>4 tools"]
      WebFetch
      WebSearch
      ToolSearch
      LSP
    Notebook["📓 Notebook<br/>1 tool"]
      NotebookEdit
    Agent["🤖 Agent<br/>3 tools"]
      Agent
      SendMessage
      TeamCreate/Delete
    Task["📋 Task<br/>6 tools"]
      TaskCreate
      TaskGet
      TaskList
      TaskUpdate
      TaskStop
      TaskOutput
    MCP["🌐 MCP<br/>3 tools"]
      MCPTool
      ListMcpResources
      ReadMcpResource
    Mode["🔄 Mode<br/>3 tools"]
      EnterPlanMode
      ExitPlanMode
      Worktree
    Schedule["⏰ Schedule<br/>4 tools"]
      CronCreate
      CronList
      CronDelete
      RemoteTrigger
    Meta["📎 Meta<br/>12+ tools"]
      Skill
      Config
      Brief
      Sleep
      AskUser

2.3.2 工具实现模式：BaseTool 抽象类

所有工具都继承自 BaseTool 抽象基类，遵循统一的实现模式：

from pydantic import BaseModel, Field
from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult

class WebSearchInput(BaseModel):
    """Pydantic 模型定义输入参数"""
    query: str = Field(
        description="搜索查询词",
        min_length=1,
        max_length=1000
    )
    num_results: int = Field(
        default=5,
        description="返回结果数量",
        ge=1,
        le=20
    )

class WebSearchTool(BaseTool):
    """Web 搜索工具实现示例"""
    
    # 工具元数据
    name = "web_search"
    description = "使用搜索引擎查询网络信息"
    input_model = WebSearchInput
    
    # 权限配置
    requires_permission = False  # 读取类操作默认无需权限
    
    async def execute(
        self,
        arguments: WebSearchInput,
        context: ToolExecutionContext
    ) -> ToolResult:
        """
        工具执行逻辑
        
        Args:
            arguments: 经过 Pydantic 验证的输入参数
            context: 执行上下文（包含配置、日志等）
        
        Returns:
            ToolResult: 标准化输出格式
        """
        try:
            # 实际搜索逻辑
            results = await self._search(
                query=arguments.query,
                num=arguments.num_results
            )
            
            return ToolResult(
                output=self._format_results(results),
                metadata={
                    "query": arguments.query,
                    "found": len(results)
                }
            )
            
        except Exception as e:
            return ToolResult(
                error=f"搜索失败: {str(e)}",
                metadata={"query": arguments.query}
            )
    
    def to_json_schema(self) -> dict:
        """
        生成 JSON Schema 供 LLM 理解工具能力
        
        OpenHarness 自动从 Pydantic 模型生成 Schema，
        确保 LLM 能够正确理解和使用工具。
        """
        return {
            "name": self.name,
            "description": self.description,
            "inputSchema": self.input_model.schema()
        }

2.3.3 Pydantic 验证与类型安全

OpenHarness 选择 Pydantic 作为输入验证框架，这是一个关键的技术决策：

优势：

1. 类型安全：利用 Python 3.11+ 的类型注解，编译时即可发现类型错误
2. 自动验证：Pydantic 自动处理类型转换和约束验证
3. JSON Schema 生成：一键生成符合 OpenAI/Anthropic 工具格式的 Schema
4. 文档生成：Pydantic 模型自带文档，提升代码可读性

对比传统字典参数：

2.3.4 工具发现与动态加载

OpenHarness 实现了工具发现机制，允许系统动态识别可用工具：

# 工具注册与发现
class ToolRegistry:
    """工具注册表，管理所有可用工具"""
    
    def __init__(self):
        self._tools: Dict[str, BaseTool] = {}
    
    def register(self, tool: BaseTool) -> None:
        """注册新工具"""
        self._tools[tool.name] = tool
    
    def get(self, name: str) -> Optional[BaseTool]:
        """按名称获取工具"""
        return self._tools.get(name)
    
    def list_all(self) -> List[BaseTool]:
        """获取所有工具"""
        return list(self._tools.values())
    
    def to_json_schemas(self) -> List[dict]:
        """生成所有工具的 JSON Schema 供 LLM 使用"""
        return [tool.to_json_schema() for tool in self._tools.values()]

这种设计使得：

• 插件可以注册新工具：Plugin 系统可以动态添加工具
• 工具可以按需加载：冷启动时只加载核心工具
• 版本管理更容易：工具可以独立演进，不影响其他组件

2.4 Skills 子系统：知识即代码

2.4.1 Skills 的本质：按需加载的领域知识

Skills（技能）是 OpenHarness 的一个创新设计。它将领域知识从代码中分离出来，以 Markdown 文件的形式存储，按需加载到 LLM 上下文中。

核心洞察：并非所有知识都需要在每次对话中加载。通过智能识别用户意图，只加载相关的技能，可以：

• 减少上下文窗口占用
• 提高响应质量（减少噪声）
• 支持动态扩展（无需修改代码）

2.4.2 Skills 文件格式

---
name: debug
version: "1.0.0"
description: 系统性诊断和修复代码缺陷
triggers:
  - "debug"
  - "fix bug"
  - "排查问题"
  - "报错"
---

# 调试技能

## 何时使用
当用户遇到代码错误、异常或行为不符合预期时使用此技能。

## 调试流程

### 1. 信息收集
首先收集关键信息：
- 错误信息和堆栈跟踪
- 相关代码片段
- 复现步骤
- 环境信息（Python版本、依赖版本）

### 2. 假设形成
基于收集的信息，形成可能的原因假设：
- 逻辑错误
- 边界条件未处理
- 依赖版本不兼容
- 并发问题

### 3. 验证与修复
逐一验证假设并实施修复：
- 添加日志或断点
- 编写最小复现案例
- 应用修复并验证

## 示例

用户："这段代码报 IndexError"

助手："我来帮您排查。首先让我查看相关代码..."
[使用 Read 工具查看代码]
"看到问题了，第15行访问列表时索引越界。建议添加边界检查..."

2.4.3 Skills 加载机制

flowchart LR
    A["用户输入"] --> B{"触发词匹配"}
    B -->|匹配成功| C["加载 Skill"]
    B -->|无匹配| D["基础模式"]
    C --> E["解析 Frontmatter"]
    E --> F["提取触发词/描述"]
    F --> G["注入系统提示"]
    G --> H["增强的 Agent Loop"]
    D --> H

实现亮点：

1. YAML Frontmatter：使用标准的 YAML 前置元数据格式，便于解析和工具处理
2. 触发词机制：通过关键词匹配自动激活技能，无需用户手动选择
3. 层级结构：支持技能继承和组合，复杂场景可以组合多个技能
4. 兼容生态：与 anthropics/skills 格式兼容，可直接使用社区技能库

2.4.4 Skills 与 Fine-tuning 的对比

OpenHarness 的选择：采用 Skills 机制而非 Fine-tuning，强调可解释性和灵活性，适合需要频繁更新知识的场景。

2.5 Plugins 子系统：扩展性架构

2.5.1 Plugin 架构设计

OpenHarness 的 Plugin 系统与 claude-code/plugins 生态兼容，允许第三方开发者扩展框架功能。

flowchart TB
    subgraph "Plugin 结构"
        PJ["plugin.json<br/>元数据"]
        CMD["commands/<br/>斜杠命令"]
        HK["hooks/<br/>生命周期钩子"]
        AGT["agents/<br/>Agent 定义"]
        MCP["mcp/<br/>MCP 服务器"]
    end
    
    subgraph "OpenHarness 集成"
        REG["Plugin Registry"]
        LD["Plugin Loader"]
        HKM["Hook Manager"]
    end
    
    PJ --> REG
    CMD --> LD
    HK --> HKM
    AGT --> REG
    MCP --> REG

2.5.2 Plugin 类型详解

1. Commands（命令插件）

添加新的斜杠命令，扩展交互能力：

<!-- commands/commit.md -->
---
name: commit
description: 创建规范的 Git 提交
---

# /commit 命令

根据当前更改生成规范的提交信息。

## 使用方式

/commit [类型]

## 参数
- `type`: 提交类型（feat|fix|docs|style|refactor|test|chore）

2. Hooks（钩子插件）

介入工具执行生命周期：

// hooks/hooks.json
{
  "pre_tool_use": [
    {
      "pattern": "Write|Edit",
      "script": "check_sensitive_files.py",
      "description": "检查是否尝试修改敏感文件"
    }
  ],
  "post_tool_use": [
    {
      "pattern": "*",
      "script": "log_usage.py",
      "description": "记录工具使用日志"
    }
  ]
}

3. Agents（Agent 插件）

定义专用的子 Agent：

<!-- agents/code-reviewer.md -->
---
name: code-reviewer
description: 专业的代码审查 Agent
model: claude-3-5-sonnet
system_prompt: |
  你是一位资深的代码审查专家。请重点关注：
  1. 潜在的 bug 和逻辑错误
  2. 性能瓶颈
  3. 安全漏洞
  4. 代码风格问题
---

4. MCP Servers（MCP 插件）

集成外部工具服务：

// mcp/config.json
{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
      }
    }
  }
}

2.5.3 已验证的插件生态

OpenHarness 官方测试了 12 个 claude-code 官方插件：

这种兼容性证明了 OpenHarness 架构设计的合理性，用户可以无缝迁移现有插件。

2.6 Permissions 子系统：多层次安全模型

2.6.1 权限模型设计

OpenHarness 实现了多层次权限控制，平衡了安全性与便利性：

flowchart TD
    A["工具调用请求"] --> B{"权限模式检查"}
    
    B -->|Auto 模式| C["自动允许"]
    B -->|Plan 模式| D["拒绝写入"]
    B -->|Default 模式| E{"路径规则检查"}
    
    E -->|匹配拒绝规则| F["拒绝执行"]
    E -->|匹配允许规则| G["允许执行"]
    E -->|无匹配规则| H{"危险命令检查"}
    
    H -->|匹配拒绝列表| F
    H -->|安全命令| I["交互式确认"]
    
    C --> J["执行工具"]
    G --> J
    I -->|用户确认| J
    I -->|用户拒绝| K["取消执行"]
    F --> K
    D --> L["返回错误信息"]

2.6.2 三种权限模式

2.6.3 路径级规则配置

{
  "permission": {
    "mode": "default",
    "path_rules": [
      {"pattern": "/etc/*", "allow": false, "reason": "系统目录保护"},
      {"pattern": "~/.ssh/*", "allow": false, "reason": "SSH 密钥保护"},
      {"pattern": "~/projects/*", "allow": true, "reason": "项目目录"},
      {"pattern": "*/node_modules/*", "allow": true, "auto": true, "reason": "依赖目录自动允许"}
    ],
    "denied_commands": [
      "rm -rf /",
      "rm -rf ~",
      "DROP TABLE",
      "DELETE FROM"
    ],
    "auto_approve_patterns": [
      "Read",
      "Glob",
      "Grep"
    ]
  }
}

设计亮点：

1. 模式匹配：支持 Glob 模式，灵活定义路径规则
2. 优先级机制：前面的规则覆盖后面的规则
3. 原因说明：每个规则都可以附带说明，帮助用户理解决策
4. 自动批准：只读类工具可以配置为自动通过

2.7 其他子系统概览

2.7.1 MCP 子系统：标准化工具协议

Model Context Protocol (MCP) 是 Anthropic 提出的开放标准，用于统一 LLM 与外部工具的交互方式。

OpenHarness 的 MCP 子系统实现了 MCP 客户端，可以连接任意 MCP 服务器：

# MCP 工具调用示例
async def call_mcp_tool(
    server_name: str,
    tool_name: str,
    arguments: dict
) -> ToolResult:
    """调用外部 MCP 服务器的工具"""
    server = mcp_registry.get_server(server_name)
    
    # 通过 stdio 或 HTTP 与 MCP 服务器通信
    result = await server.call_tool(tool_name, arguments)
    
    return ToolResult(output=result.content)

价值：通过支持 MCP，OpenHarness 可以接入不断增长的 MCP 生态，无需为每个新工具编写适配代码。

2.7.2 Memory 子系统：持久化记忆

Memory 子系统负责跨会话的知识持久化：

class MemoryManager:
    """记忆管理器"""
    
    async def store(
        self,
        key: str,
        value: Any,
        ttl: Optional[int] = None
    ) -> None:
        """存储记忆项"""
        await self.db.execute(
            "INSERT INTO memory (key, value, expires_at) VALUES (?, ?, ?)",
            (key, json.dumps(value), 
             datetime.now() + timedelta(seconds=ttl) if ttl else None)
        )
    
    async def retrieve(self, key: str) -> Optional[Any]:
        """检索记忆项"""
        row = await self.db.fetchone(
            "SELECT value FROM memory WHERE key = ? AND (expires_at IS NULL OR expires_at > ?)",
            (key, datetime.now())
        )
        return json.loads(row[0]) if row else None

应用场景：

• 跨会话的代码库理解
• 用户偏好记忆
• 长期任务状态跟踪

2.7.3 Tasks 子系统：后台任务管理

Tasks 子系统支持长时间运行的后台任务：

class TaskManager:
    """异步任务管理"""
    
    async def create_task(
        self,
        description: str,
        agent_config: AgentConfig
    ) -> Task:
        """创建新任务"""
        task = Task(
            id=uuid.uuid4(),
            description=description,
            status=TaskStatus.PENDING
        )
        
        # 在后台运行 Agent
        asyncio.create_task(
            self._run_agent_task(task, agent_config)
        )
        
        return task

使用场景：

• 大型代码库分析
• 批量文件处理
• 长时间运行的测试

2.8 架构设计的权衡分析

2.8.1 为什么选择 Python？

2.8.2 为什么精简企业功能？

2.8.3 架构的可扩展性评估

OpenHarness 的架构设计在以下场景表现出良好的扩展性：

✅ 容易扩展：

• 添加新工具（继承 BaseTool）
• 添加新技能（编写 .md 文件）
• 添加新插件（遵循标准格式）
• 添加新 LLM 提供商（实现抽象接口）

⚠️ 需要谨慎扩展：

• 修改 Agent Loop 核心逻辑
• 变更工具调用协议
• 替换存储后端

❌ 不建议扩展：

• 直接修改子系统内部实现
• 绕过权限检查机制

2.9 小结

OpenHarness 通过 10 个精心设计的子系统，在 11,733 行代码中实现了完整的 Agent Harness 能力。其核心架构亮点包括：

1. 清晰的职责划分：每个子系统只关注一个领域，降低认知负担
2. 强大的 Agent Loop：流式处理 + 工具调用循环，支撑复杂任务
3. 类型安全的工具系统：Pydantic + JSON Schema，确保 LLM 正确调用
4. 灵活的扩展机制：Skills、Plugins、MCP 三层扩展体系
5. 多层次安全模型：Default/Auto/Plan 三种模式 + 路径级规则

下一模块中，我们将对比分析 OpenHarness 与 Claude Code、OpenClaw、nanobot、Cursor 等竞品，帮助读者做出明智的选型决策。

参考资料

1. OpenHarness GitHub - 架构文档 - 项目官方架构说明
2. Model Context Protocol 规范 - MCP 官方文档
3. Pydantic 文档 - 输入验证框架
4. Anthropic Tool Use 指南 - LLM 工具调用最佳实践
5. Claude Code Plugins 仓库 - 插件生态参考

上一页：模块一：项目背景与目标 | 下一页：模块三：竞品对比与选型分析