OpenHarness 深度解析：当 1.1 万行 Python 把闭源巨头的 51.2 万行代码「压缩」成开源自由

4 月 2 日开源，两天登顶 GitHub Trending 全球榜首。用 Python 写了 Claude Code 核心架构的 98%，体积缩小 44 倍，完全开源免费，模型无关。这不是魔法，这是香港大学 HKUDS 团队交出的答卷。

一、从「黑盒崇拜」到「透明时代」

过去两年，AI Agent 领域有个不成文的共识：Anthropic 的 Claude Code 是行业标准。51.2 万行 TypeScript，1884 个文件，覆盖文件操作、Shell 执行、Web 搜索、MCP 协议等 44 种核心工具，深度绑定 Claude 模型生态，形成了完整的「模型 + 框架」闭环。

但这个闭环有个致命问题：闭源。

想看看工具调用是怎么实现的？不知道。想魔改记忆系统？没门。想换成 GPT 或 Kimi？对不起，只支持 Claude。想本地部署保护数据隐私？做梦。

OpenHarness 的出现，就像当年的 Linux 之于 Unix——用开源撕开了闭源的护城河。

Claude Code:     51.2 万行 TypeScript | 1884 个文件 | 闭源
OpenHarness:      1.17 万行 Python   | 163 个文件  | MIT 开源
体积差距:         44 倍
工具覆盖率:       98% (43/44)

这不是简单的「复刻」，这是架构层面的降维打击。

二、核心设计哲学：Model is the Agent, Code is the Harness

OpenHarness 的核心理念可以用一句话概括：

Agent = Model (大脑) + Harness (手脚/记忆/安全边界)

2.1 模型层（Agent）vs 框架层（Harness）

这个设计哲学解决了一个根本问题：LLM 本身不是 Agent。

原始的大语言模型（无论 GPT-4 还是 Claude 3.5）只是一个「输入文本 → 输出文本」的黑盒子。它没有手（无法执行命令）、没有记忆（每次对话独立）、没有安全边界（无法防止恶意操作）。

Harness 就是为模型装上这些能力的基础设施：

2.2 模型无关性：打破厂商锁定的钥匙

这是 OpenHarness 与 Claude Code 最本质的区别。

# config.yaml - 切换模型只需改配置
llm:
  provider: openai          # 可选 anthropic / openai / deepseek / ollama
  api_key: sk-xxxxxxxx
  model: gpt-4o             # 或 claude-3-5-sonnet / deepseek-chat / qwen-max
  base_url: https://api.openai.com/v1

这意味着：

1. 成本优化：任务简单时用便宜的模型，复杂任务切换到强大的模型
2. 隐私保护：本地部署 Ollama，数据不出机房
3. 多模型编排：不同任务用不同模型，形成「混合智能体」
4. 厂商竞争受益者：谁家模型强就用谁家，不绑定任何一家

三、架构解析：10 个子系统如何撑起智能体运行时

OpenHarness 把复杂的 Agent Harness 拆解成了 10 个边界清晰、逻辑独立的子系统。每个模块都有完整的 Pydantic 类型校验和权限检查。

3.1 整体架构分层（5 层设计）

┌─────────────────────────────────────────────────────────────┐
│                   交互层 (Interface Layer)                   │
│              CLI / React / Ink / 流式输出                    │
├─────────────────────────────────────────────────────────────┤
│                Agent 核心循环层 (Agent Core)                 │
│        AgentLoop / Planner / PromptEngine / StateMachine    │
├─────────────────────────────────────────────────────────────┤
│                 能力扩展层 (Capability Layer)                │
│    ToolSystem / Memory / Security / MultiAgent / Skills     │
├─────────────────────────────────────────────────────────────┤
│                执行引擎层 (Execution Engine)                 │
│     CommandExecutor / FileIOEngine / Sandbox / ToolInvoker  │
├─────────────────────────────────────────────────────────────┤
│                基础支撑层 (Infrastructure)                   │
│     Config / Logging / TokenCounter / LLMAdapter / Hooks    │
└─────────────────────────────────────────────────────────────┘

3.2 子系统详解

3.2.1 Agent Loop（核心循环）—— 智能体的心脏

这是整个框架的心脏，实现了标准的 ReAct (Reason + Act) 流程：

# 核心循环伪代码
class AgentLoop:
    async def run(self, user_input: str) -> Response:
        state = State.IDLE
        
        while state != State.DONE:
            # 1. 思考：理解意图，规划下一步
            thought = await self.think(user_input)
            
            # 2. 决策：选择工具和参数
            action = await self.decide(thought)
            
            # 3. 执行：调用工具
            result = await self.execute(action)
            
            # 4. 观察：评估结果
            observation = await self.observe(result)
            
            # 5. 反思：是否需要继续
            state = self.reflect(observation)
            
        return self.summarize()

流式工具调用是 OpenHarness 的一大亮点：边思考边执行，而不是等完整规划后再执行。这显著降低了首字延迟，提升了用户体验。

3.2.2 工具系统—— 43 种内置工具

工具是智能体的「手」，OpenHarness 内置了 43 种核心工具：

# 工具统一抽象接口
class BaseTool(ABC):
    @abstractmethod
    def run(self, parameters: dict) -> ToolResult:
        """执行工具，返回结果"""
        pass
    
    @abstractmethod
    def validate(self, parameters: dict) -> bool:
        """参数校验（Pydantic 自动生成）"""
        pass
    
    @abstractmethod
    def check_permission(self, parameters: dict) -> bool:
        """权限检查"""
        pass

工具分类：

示例：文件读取工具的实现

class ReadFileTool(BaseTool):
    name = "read_file"
    description = "读取文件内容，支持指定编码和行范围"
    
    parameters_schema = {
        "path": {"type": "string", "required": True},
        "encoding": {"type": "string", "default": "utf-8"},
        "start_line": {"type": "integer", "default": 1},
        "end_line": {"type": "integer", "default": None}
    }
    
    def check_permission(self, params: dict) -> bool:
        path = params["path"]
        # 检查是否在白名单内
        return self.security.is_path_allowed(path)
    
    def run(self, params: dict) -> ToolResult:
        path = params["path"]
        encoding = params.get("encoding", "utf-8")
        
        try:
            with open(path, 'r', encoding=encoding) as f:
                lines = f.readlines()
            
            start = params.get("start_line", 1) - 1
            end = params.get("end_line", len(lines))
            
            content = ''.join(lines[start:end])
            return ToolResult(success=True, output=content)
            
        except Exception as e:
            return ToolResult(success=False, error=str(e))

3.2.3 记忆系统—— 突破 Token 限制的关键

LLM 的原生上下文窗口有限（通常 4k-200k tokens），而复杂任务往往需要跨会话的「长期记忆」。OpenHarness 提供了三层记忆架构：

class MemorySystem:
    def __init__(self):
        self.working_memory = WorkingMemory()      # 短期：当前对话上下文
        self.persistent_memory = PersistentMemory() # 长期：跨会话存储
        self.vector_memory = VectorMemory()         # 检索：语义搜索
    
    async def remember(self, key: str, value: Any, ttl: int = None):
        """存储记忆"""
        # 短期记忆
        self.working_memory.set(key, value)
        
        # 长期记忆（可选）
        if ttl is None:  # 永久存储
            await self.persistent_memory.save(key, value)
    
    async def recall(self, query: str, top_k: int = 5) -> list:
        """检索记忆"""
        # 语义搜索
        results = await self.vector_memory.search(query, top_k)
        return results

实际应用场景：

# 场景：让 Agent 记住用户的代码风格偏好
await memory.remember("user_coding_style", {
    "language": "Python",
    "style": "functional",
    "formatter": "black",
    "type_hints": True
})

# 后续对话中自动应用
style = await memory.recall("coding style preferences")
# → 自动用 functional 风格生成代码，自动添加 type hints

3.2.4 安全与治理—— 防止 Agent 「失控」

这是 Agent 落地最关键的问题：如何防止 AI 执行危险操作？

OpenHarness 实现了多级安全控制：

class SecurityGovernor:
    def __init__(self, config: SecurityConfig):
        self.path_whitelist = config.path_whitelist      # 文件路径白名单
        self.command_blacklist = config.command_blacklist # 命令黑名单
        self.sandbox_enabled = config.sandbox_enabled     # 沙箱模式
        
    def check_file_access(self, path: str, mode: str) -> bool:
        """检查文件访问权限"""
        abs_path = os.path.abspath(path)
        
        # 白名单检查
        if not any(abs_path.startswith(allowed) for allowed in self.path_whitelist):
            return False
            
        # 敏感路径保护
        if any(p in abs_path for p in [".ssh", ".env", "credentials"]):
            return False
            
        return True
    
    def check_command(self, command: str) -> bool:
        """检查命令执行权限"""
        # 黑名单检查
        dangerous_commands = ["rm -rf", "sudo", "chmod 777", "dd if="]
        if any(danger in command for danger in dangerous_commands):
            return False
            
        return True

沙箱隔离模式：

# 启用沙箱模式后，所有危险操作都在隔离环境中执行
class Sandbox:
    def execute(self, command: str) -> SandboxResult:
        # 1. 创建隔离环境
        with tempfile.TemporaryDirectory() as sandbox_dir:
            # 2. 挂载必要的文件（只读）
            self.mount_readonly(sandbox_dir, "/usr/lib")
            
            # 3. 执行命令
            result = subprocess.run(
                command,
                cwd=sandbox_dir,
                capture_output=True,
                timeout=30
            )
            
            # 4. 清理环境
            # sandbox_dir 自动删除
            
        return SandboxResult(
            stdout=result.stdout,
            stderr=result.stderr,
            exit_code=result.returncode
        )

3.2.5 可观测性—— 调试 Agent 的「黑盒」

Agent 的行为往往难以预测，可观测性是调试的关键：

class Observability:
    def __init__(self):
        self.logger = Logger()
        self.token_counter = TokenCounter()
        self.cost_tracker = CostTracker()
        
    def trace_tool_call(self, tool: str, params: dict, result: ToolResult):
        """记录工具调用"""
        self.logger.info(f"Tool: {tool}", {
            "params": params,
            "success": result.success,
            "duration_ms": result.duration_ms,
            "tokens_used": result.tokens_used
        })
        
    def track_cost(self, model: str, input_tokens: int, output_tokens: int):
        """追踪成本"""
        cost = self.calculate_cost(model, input_tokens, output_tokens)
        self.cost_tracker.add(cost)
        
    def get_summary(self) -> dict:
        """获取运行摘要"""
        return {
            "total_tokens": self.token_counter.total,
            "total_cost": self.cost_tracker.total,
            "tool_calls": self.logger.count("Tool:"),
            "errors": self.logger.count("ERROR")
        }

3.2.6 多智能体协同—— 从单体到分布式

复杂任务往往需要多个专业智能体协作完成：

class MultiAgentCoordinator:
    def __init__(self):
        self.agents = {}
        self.message_bus = MessageBus()
        
    def register_agent(self, name: str, agent: Agent, role: str):
        """注册智能体"""
        self.agents[name] = {
            "agent": agent,
            "role": role,
            "status": "idle"
        }
        
    async def dispatch(self, task: Task) -> Result:
        """任务分发"""
        # 1. 分析任务，确定需要的智能体
        required_roles = self.analyze_task(task)
        
        # 2. 分配任务
        subtasks = self.split_task(task, required_roles)
        
        # 3. 并行执行
        results = await asyncio.gather(*[
            self.agents[role]["agent"].run(subtask)
            for role, subtask in subtasks.items()
        ])
        
        # 4. 合并结果
        return self.merge_results(results)

协作示例：

# 场景：开发一个 Web 应用
coordinator = MultiAgentCoordinator()

# 注册专业智能体
coordinator.register_agent("architect", ArchitectAgent(), "架构设计")
coordinator.register_agent("frontend", FrontendAgent(), "前端开发")
coordinator.register_agent("backend", BackendAgent(), "后端开发")
coordinator.register_agent("tester", TesterAgent(), "测试验证")

# 执行任务
result = await coordinator.dispatch(Task(
    description="开发一个 Todo 应用",
    requirements=["React 前端", "FastAPI 后端", "PostgreSQL 数据库"]
))

3.3 与 Claude Code 的架构对比

四、代码实战：从零构建一个智能体

4.1 快速开始：5 分钟搭建你的第一个 Agent

# 1. 克隆项目
git clone https://github.com/HKUDS/OpenHarness.git
cd OpenHarness

# 2. 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 3. 安装依赖
pip install -r requirements.txt

# 4. 配置模型
cp config.example.yaml config.yaml
# 编辑 config.yaml，填入 API Key

# 5. 启动
python main.py

4.2 配置详解：多模型接入

# config.yaml
llm:
  # 主模型
  primary:
    provider: anthropic
    api_key: ${ANTHROPIC_API_KEY}  # 支持环境变量
    model: claude-sonnet-4-20250514
    max_tokens: 4096
    
  # 备用模型（成本优化）
  fallback:
    provider: openai
    api_key: ${OPENAI_API_KEY}
    model: gpt-4o-mini
    
  # 本地模型（隐私保护）
  local:
    provider: ollama
    base_url: http://localhost:11434
    model: llama3.1:8b

# 安全配置
security:
  path_whitelist:
    - /Users/yourname/projects
    - /tmp/agent_workspace
  command_blacklist:
    - rm -rf
    - sudo
    - chmod 777
  sandbox_enabled: true
  
# 记忆配置
memory:
  persistent:
    enabled: true
    storage_path: ~/.openharness/memory
  vector:
    enabled: true
    embedding_model: text-embedding-3-small

4.3 自定义工具开发

# custom_tools/github_tool.py
from openharness import BaseTool, ToolResult
import requests

class GitHubSearchTool(BaseTool):
    """GitHub 代码搜索工具"""
    
    name = "github_search"
    description = "在 GitHub 上搜索代码、仓库或 issues"
    
    parameters_schema = {
        "query": {
            "type": "string",
            "description": "搜索关键词",
            "required": True
        },
        "type": {
            "type": "string",
            "enum": ["code", "repositories", "issues"],
            "default": "code"
        },
        "language": {
            "type": "string",
            "description": "编程语言过滤",
            "default": None
        }
    }
    
    def __init__(self, github_token: str):
        self.github_token = github_token
        self.api_base = "https://api.github.com"
        
    def run(self, params: dict) -> ToolResult:
        query = params["query"]
        search_type = params.get("type", "code")
        language = params.get("language")
        
        # 构建搜索 URL
        url = f"{self.api_base}/search/{search_type}"
        
        # 构建查询
        q = query
        if language:
            q += f" language:{language}"
            
        # 发送请求
        headers = {
            "Authorization": f"token {self.github_token}",
            "Accept": "application/vnd.github.v3+json"
        }
        
        response = requests.get(
            url,
            params={"q": q, "per_page": 10},
            headers=headers
        )
        
        if response.status_code != 200:
            return ToolResult(
                success=False,
                error=f"GitHub API 错误: {response.status_code}"
            )
            
        data = response.json()
        results = []
        
        for item in data.get("items", [])[:5]:
            if search_type == "code":
                results.append({
                    "file": item["name"],
                    "repository": item["repository"]["full_name"],
                    "url": item["html_url"],
                    "score": item.get("score", 0)
                })
            else:
                results.append({
                    "name": item["name"],
                    "url": item["html_url"],
                    "description": item.get("description", ""),
                    "stars": item.get("stargazers_count", 0)
                })
                
        return ToolResult(
            success=True,
            output={
                "total": data.get("total_count", 0),
                "results": results
            }
        )

4.4 注册自定义工具

# main.py
from openharness import Agent, ToolRegistry
from custom_tools.github_tool import GitHubSearchTool

# 创建 Agent
agent = Agent(config_path="config.yaml")

# 注册自定义工具
agent.register_tool(GitHubSearchTool(
    github_token=os.environ.get("GITHUB_TOKEN")
))

# 运行
agent.run()

4.5 多智能体协作示例

# multi_agent_demo.py
from openharness import MultiAgentCoordinator, Agent

# 创建协调器
coordinator = MultiAgentCoordinator()

# 研究智能体
researcher = Agent(
    name="researcher",
    system_prompt="""你是一个研究助手，负责：
    1. 搜索和分析技术文档
    2. 提取关键信息
    3. 生成研究报告""",
    tools=["web_search", "read_file", "parse_pdf"]
)

# 编码智能体
coder = Agent(
    name="coder",
    system_prompt="""你是一个编程助手，负责：
    1. 编写高质量代码
    2. 实现研究智能体提出的技术方案
    3. 编写单元测试""",
    tools=["write_file", "read_file", "run_python", "execute_command"]
)

# 审查智能体
reviewer = Agent(
    name="reviewer",
    system_prompt="""你是一个代码审查助手，负责：
    1. 检查代码质量
    2. 发现潜在问题
    3. 提出改进建议""",
    tools=["read_file", "run_python"]
)

# 注册智能体
coordinator.register_agent("researcher", researcher)
coordinator.register_agent("coder", coder)
coordinator.register_agent("reviewer", reviewer)

# 执行任务
result = await coordinator.run("""
开发一个 Python 脚本，实现以下功能：
1. 从 GitHub API 获取指定用户的仓库列表
2. 统计各仓库的 star 数和 fork 数
3. 生成可视化图表
4. 输出 Markdown 报告
""")

五、性能优化：让 Agent 跑得更快

5.1 流式工具调用

传统 Agent 的执行流程是串行的：

思考 → 规划 → 执行工具1 → 等待 → 执行工具2 → 等待 → 总结

OpenHarness 支持流式工具调用：

# 流式执行：边思考边执行
async def stream_execute(self, tasks: list[Task]):
    results = []
    for task in tasks:
        # 不等待上一个任务完成，立即发起下一个
        future = self.execute_async(task)
        results.append(future)
    
    # 并行收集结果
    return await asyncio.gather(*results)

5.2 Token 优化

class TokenOptimizer:
    """Token 使用优化器"""
    
    def compress_context(self, messages: list[dict]) -> list[dict]:
        """压缩上下文"""
        # 1. 移除冗余信息
        messages = self.remove_duplicates(messages)
        
        # 2. 摘要旧消息
        if len(messages) > 20:
            old_messages = messages[:-10]
            summary = self.summarize(old_messages)
            messages = [{"role": "system", "content": summary}] + messages[-10:]
            
        # 3. 截断过长内容
        messages = self.truncate_long_content(messages, max_length=2000)
        
        return messages
    
    def estimate_tokens(self, text: str) -> int:
        """估算 Token 数量"""
        # 简单估算：中文约 1.5 字/token，英文约 4 字/token
        chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
        other_chars = len(text) - chinese_chars
        return int(chinese_chars / 1.5 + other_chars / 4)

5.3 缓存策略

class ToolResultCache:
    """工具结果缓存"""
    
    def __init__(self, ttl: int = 3600):
        self.cache = {}
        self.ttl = ttl
        
    def get_cache_key(self, tool: str, params: dict) -> str:
        """生成缓存键"""
        return hashlib.md5(
            f"{tool}:{json.dumps(params, sort_keys=True)}".encode()
        ).hexdigest()
    
    async def get_or_execute(self, tool: BaseTool, params: dict) -> ToolResult:
        """获取缓存或执行"""
        key = self.get_cache_key(tool.name, params)
        
        if key in self.cache:
            cached = self.cache[key]
            if time.time() - cached["timestamp"] < self.ttl:
                return cached["result"]
        
        # 执行并缓存
        result = await tool.run_async(params)
        self.cache[key] = {
            "result": result,
            "timestamp": time.time()
        }
        return result

六、实战场景：OpenHarness 的典型应用

6.1 自动化代码审查

# code_reviewer.py
from openharness import Agent

reviewer = Agent(
    system_prompt="""你是一个资深代码审查专家，负责：
    1. 检查代码风格是否符合 PEP 8 / Google Style
    2. 发现潜在的 bug 和安全问题
    3. 评估代码可维护性
    4. 提供具体的改进建议""",
    tools=["read_file", "run_python", "execute_command"]
)

# 审查整个项目
result = reviewer.run("""
审查 ./src 目录下的所有 Python 文件：
1. 检查代码风格（使用 black 和 ruff）
2. 运行单元测试
3. 生成审查报告
""")

6.2 技术文档生成

# doc_generator.py
from openharness import Agent

doc_writer = Agent(
    system_prompt="""你是一个技术文档撰写专家，负责：
    1. 分析代码结构
    2. 提取 API 接口
    3. 生成清晰的文档""",
    tools=["read_file", "write_file", "execute_command"]
)

result = doc_writer.run("""
为 ./src/api 目录生成 API 文档：
1. 分析所有路由和处理器
2. 提取请求/响应格式
3. 生成 OpenAPI 规范
4. 生成 Markdown 文档
""")

6.3 智能运维助手

# ops_assistant.py
from openharness import Agent

ops = Agent(
    system_prompt="""你是一个运维专家，负责：
    1. 监控系统状态
    2. 分析日志
    3. 诊断问题
    4. 执行修复操作""",
    tools=["execute_command", "read_file", "web_search"]
)

# 启用严格的权限控制
ops.config.security.path_whitelist = ["/var/log", "/etc/nginx"]
ops.config.security.command_whitelist = [
    "systemctl status *",
    "tail -n 100 *",
    "nginx -t"
]

result = ops.run("""
检查 Nginx 服务状态，如果发现异常：
1. 分析最近 100 条错误日志
2. 搜索常见解决方案
3. 尝试修复问题
""")

6.4 数据分析助手

# data_analyst.py
from openharness import Agent

analyst = Agent(
    system_prompt="""你是一个数据分析专家，负责：
    1. 数据清洗和预处理
    2. 统计分析
    3. 可视化生成
    4. 报告撰写""",
    tools=["read_file", "write_file", "run_python", "execute_command"]
)

result = analyst.run("""
分析 ./data/sales.csv：
1. 数据清洗（处理缺失值和异常值）
2. 计算关键指标（GMV、客单价、复购率）
3. 生成可视化图表
4. 输出分析报告
""")

七、与 Claude Code 的深度对比

7.1 功能覆盖

7.2 适用人群

选择 Claude Code 如果你：

• 是专业开发者，需要开箱即用的稳定体验
• 深度使用 Claude 模型，追求最佳协同效果
• 需要企业级功能、安全合规、官方技术支持
• 不想折腾，专注开发本身

选择 OpenHarness 如果你：

• 是研究者/学生，想学习 Agent 底层原理
• 需要高度定制，或接入多种模型
• 关注数据隐私，需要本地部署
• 想二次开发，构建自己的 Agent 产品

7.3 成本对比

Claude Code:
- 订阅费: $20/月 (Claude Pro)
- API 调用: 按 Token 计费
- 闭源: 无法自托管

OpenHarness:
- 软件费: $0 (MIT 开源)
- API 调用: 可选择任意模型（包括免费模型）
- 自托管: 完全本地运行，无额外费用

八、局限性与未来展望

8.1 当前局限性

1. 生态不如 Claude Code 成熟

   • 插件数量较少
   • 社区规模有限

2. 安全机制相对基础

   • 沙箱隔离不如 Claude Code 完善
   • 缺少企业级审计日志

3. 长上下文优化不足

   • 200k+ token 场景下性能不如 Claude Code

4. IDE 集成缺失

   • 没有 VS Code / JetBrains 插件

8.2 未来发展方向

1. 更丰富的工具生态

   • 支持更多 MCP 服务器
   • 社区贡献的工具库

2. 企业级功能增强

   • 完善的 RBAC 权限系统
   • 审计日志和合规报告

3. 多模态能力

   • 图像理解
   • 音频处理
   • 视频分析

4. 分布式执行

   • 跨机器任务分发
   • 负载均衡和容错

九、总结：开源的意义

OpenHarness 的价值不仅仅是「复刻了 Claude Code」，更在于：

1. 透明性：你可以看到每一行代码如何运作
2. 可控性：你可以修改任何模块适应需求
3. 自由性：你可以使用任何模型，部署在任何地方
4. 教育性：它是学习 Agent 架构的最佳教材

用一句话总结：

Claude Code 是最好的商业 Agent 产品，OpenHarness 是最好的 Agent 学习平台。
干活用前者，研究用后者。

项目地址：https://github.com/HKUDS/OpenHarness
Star 数：7000+（截至 2026-04-08）
协议：MIT License

本文由程序员茄子原创发布，技术细节基于 OpenHarness 官方文档和源码分析。如有疏漏，欢迎指正。