大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



Github 地址 版本: v0.1.6

OpenHarness (简称 oh) 是一个开源的 Python Agent 基础设施框架，实现了完整的 Agent Harness 模式——即包裹 LLM 使其成为功能性 Agent 所需的全部基础设施：工具调用、技能系统、记忆、权限治理和多 Agent 协调。

核心理念："The model is the agent. The code is the harness." 模型提供智能，Harness 提供手、眼、记忆和安全边界。

ohmo 是构建在 OpenHarness 之上的个人 AI Agent 应用，通过飞书/Slack/Telegram/Discord 等消息渠道提供长期运行的助手服务，复用用户现有的 Claude Code 或 Codex 订阅。

---

OpenHarness-main/ ├── src/openharness/ # 核心库（250+ 文件） │ ├── api/ # LLM API 客户端层 │ ├── auth/ # 认证管理（API Key / OAuth / 外部凭据绑定） │ ├── bridge/ # Bridge 层（会话运行器、Work Secret） │ ├── channels/ # 消息渠道适配器 │ │ ├── bus/ # 消息总线（事件/队列） │ │ └── impl/ # 各渠道实现（12 种） │ ├── commands/ # 斜杠命令注册表（54 条命令） │ ├── config/ # 配置系统（多层级设置、路径管理） │ ├── coordinator/ # 多 Agent 协调器 │ ├── engine/ # 核心 Agent Loop 引擎 │ ├── hooks/ # 生命周期钩子（PreToolUse / PostToolUse） │ ├── keybindings/ # 键绑定系统 │ ├── mcp/ # Model Context Protocol 客户端 │ ├── memory/ # 持久化记忆系统 │ ├── output_styles/ # 输出样式加载 │ ├── permissions/ # 权限检查系统 │ ├── personalization/ # 个性化特征提取 │ ├── plugins/ # 插件系统 │ ├── prompts/ # 系统提示词构建 │ ├── sandbox/ # 沙箱运行时适配 │ ├── services/ # 服务层 │ │ ├── compact/ # 上下文自动压缩（56KB，最大单文件） │ │ ├── lsp/ # LSP 语言服务集成 │ │ └── oauth/ # OAuth 集成 │ ├── skills/ # 技能系统（Markdown 知识加载） │ ├── state/ # 应用状态管理 │ ├── swarm/ # Swarm 多 Agent 后端 │ ├── tasks/ # 后台任务管理 │ ├── themes/ # TUI 主题 │ ├── tools/ # 43+ 工具实现 │ ├── types/ # 共享类型定义 │ ├── ui/ # UI 层（React TUI 后端 + 前端启动器） │ ├── utils/ # 工具函数 │ ├── vim/ # Vim 模式 │ ├── voice/ # 语音功能 │ ├── cli.py # CLI 入口（Typer，48KB） │ └── platforms.py # 平台检测 │ ├── ohmo/ # ohmo 个人 Agent 应用 │ ├── cli.py # ohmo CLI 入口 │ ├── runtime.py # ohmo 运行时 │ ├── prompts.py # ohmo 系统提示词构建 │ ├── workspace.py # 工作空间管理（~/.ohmo/） │ ├── memory.py # ohmo 记忆管理 │ ├── session_storage.py # 会话存储后端 │ └── gateway/ # 网关服务 │ ├── bridge.py # 消息桥接 │ ├── config.py # 网关配置 │ ├── models.py # 数据模型 │ ├── router.py # 路由 │ ├── runtime.py # 会话运行时池 │ └── service.py # 网关服务生命周期 │ ├── frontend/terminal/ # React/Ink TUI 前端 │ ├── src/ │ │ ├── App.tsx # 主应用组件 │ │ ├── components/ # UI 组件（16 个） │ │ ├── hooks/ # 自定义 Hook │ │ └── theme/ # 主题系统 │ └── package.json # 依赖: ink, react, marked │ ├── tests/ # 测试套件（117 文件） ├── scripts/ # 脚本（安装、E2E 测试） ├── docs/ # 文档 ├── .agents/ # Agent 定义（空） ├── .claude/skills/ # Claude 技能（harness-eval, pr-merge） ├── pyproject.toml # 项目配置 └── README.md # 项目说明 

---

┌───────────────────────────────────────────────────────────┐ │ 用户交互层 │ │ CLI (typer) │ React/Ink TUI │ Gateway (ohmo) │ ├───────────────────────────────────────────────────────────┤ │ Runtime 组装层 │ │ RuntimeBundle (ui/runtime.py) │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐ │ │ │ QueryEngine│ │ ToolRegistry│ │ HookExecutor│ │ McpClient │ │ │ └──────────┘ └──────────┘ └──────────┘ └─────────────┘ │ ├───────────────────────────────────────────────────────────┤ │ 核心 Agent Loop │ │ engine/query.py │ │ ┌────────────────────────────────────────────────────┐ │ │ │ LLM API → 响应解析 → 工具调用 → 权限检查 → │ │ │ │ Hook 执行 → 工具执行 → 结果收集 → 循环 │ │ │ └────────────────────────────────────────────────────┘ │ ├───────────────────────────────────────────────────────────┤ │ 10 大子系统 │ │ engine/ tools/ skills/ plugins/ permissions/ │ │ hooks/ commands/ mcp/ memory/ tasks/ │ │ coordinator/ prompts/ config/ channels/ services/ │ ├───────────────────────────────────────────────────────────┤ │ API 客户端层 │ │ AnthropicApiClient │ OpenAICompatibleClient │ │ │ CopilotClient │ CodexApiClient │ │ ├───────────────────────────────────────────────────────────┤ │ 认证管理层 │ │ AuthManager │ ProviderProfile │ ExternalBinding │ │ ├───────────────────────────────────────────────────────────┤ │ 数据持久层 │ │ ~/.openharness/ │ │ ├── settings.json (全局配置) │ │ ├── credentials.json (凭据存储) │ │ ├── skills/ (用户技能) │ │ ├── plugins/ (用户插件) │ │ ├── sessions/ (会话历史) │ │ └── memory/ (持久记忆) │ └───────────────────────────────────────────────────────────┘ 

---

Agent Loop 是 OpenHarness 的心脏，实现了完整的查询-工具调用循环。

核心文件：

Agent Loop 核心流程：

# engine/query.py — run_query() 伪代码 while max_turns not exceeded: # 1. 自动压缩检查 await auto_compact_if_needed(messages) # 2. 流式调用 LLM API async for event in api_client.stream_message(request): yield text_delta_events # 3. 解析 LLM 响应 if stop_reason != "tool_use": break # 模型完成，退出循环 # 4. 执行工具调用 for tool_call in response.tool_uses: # 4a. PreHook 检查 pre_hooks = await hook_executor.execute(PRE_TOOL_USE, ...) if pre_hooks.blocked: skip # 4b. 权限检查 decision = permission_checker.evaluate(tool_name, ...) if not decision.allowed: deny # 4c. 执行工具 result = await tool.execute(parsed_input, context) # 4d. PostHook 通知 await hook_executor.execute(POST_TOOL_USE, ...) # 4e. 记录工具状态（carry-over metadata） _record_tool_carryover(...) # 5. 追加工具结果到消息列表，继续循环 messages.append(tool_results) 

关键设计决策：

• 单工具顺序执行，多工具并行执行：当 LLM 返回 1 个工具调用时顺序执行；多个工具调用时使用 asyncio.gather 并行执行
• 自动压缩（Auto-Compact）：每轮开始前检查 token 估算，先尝试 microcompact（清除旧工具结果），不够则 LLM-based 摘要
• Reactive Compact：遇到 "prompt too long" 错误时自动触发压缩并重试
• 工具状态 carry-over：通过 tool_metadata 跨压缩保持关键状态（用户目标、活跃文件、工作日志等）

QueryEngine（engine/query_engine.py）是更高层的封装：

• 管理对话历史 self._messages
• 封装 QueryContext 构建
• 提供 submit_message() / continue_pending() API
• 集成 CostTracker 统计用量
• 支持 Coordinator 上下文注入

基础抽象（tools/base.py）：

class BaseTool(ABC): name: str # 工具名称 description: str # 工具描述 input_model: type[BaseModel] # Pydantic 输入模型 async def execute(self, arguments, context: ToolExecutionContext) -> ToolResult def is_read_only(self, arguments) -> bool # 是否只读 def to_api_schema(self) -> dict # 生成 LLM 可理解的 JSON Schema class ToolRegistry: def register(tool: BaseTool) -> None def get(name: str) -> BaseTool | None def to_api_schema(self) -> list[dict] # 所有工具的 API Schema 

43+ 内置工具按类别：

工具执行管线：

LLM 返回 tool_use → Pydantic 输入验证 (model_validate) → 路径规范化 (resolve permission file path) → 权限检查 (PermissionChecker.evaluate) → 可能需要用户确认 (permission_prompt) → 执行工具 (tool.execute) → 记录 carry-over metadata → PostHook 通知 → 返回 ToolResultBlock 

技能是按需加载的知识，以 Markdown 文件存储，LLM 通过 skill 工具触发加载。

技能加载流程（skills/loader.py）：

1. 加载 bundled 内置技能（7 个：commit, debug, diagnose, plan, review, simplify, test）
2. 加载 ~/.openharness/skills/ 用户技能
3. 加载插件提供的技能

技能文件格式（SKILL.md）：

--- name: my-skill description: Expert guidance for my specific domain --- # My Skill  When to use Use when the user asks about [your domain].  Workflow 1. Step one 2. Step two 

• 支持 YAML frontmatter（通过 yaml.safe_load 解析）
• 兼容 anthropics/skills 格式
• 只需将 .md 文件放入 ~/.openharness/skills/ /SKILL.md

兼容 claude-code plugins 格式。已测试 12 个官方插件。

插件结构：

.openapi/plugins/my-plugin/.claude-plugin/ ├── plugin.json # 插件清单 ├── commands/*.md # 斜杠命令定义 ├── hooks/hooks.json # 钩子定义 └── agents/*.md # Agent 定义 

插件加载（plugins/loader.py，22KB）：

1. 扫描项目目录 .openharness/plugins/
2. 扫描用户目录 ~/.openharness/plugins/
3. 解析 plugin.json 清单
4. 加载 commands / hooks / agents / skills / MCP servers
5. 按插件 enabled/disabled 状态过滤

CLI 管理：

oh plugin list # 列出已安装插件 oh plugin install 
     
    
        
        # 从路径安装 oh plugin 
        enable 
         
         # 启用插件 
         
       

三级权限模式：

PermissionChecker 评估管线（permissions/checker.py）：

1. 内置敏感路径保护（永远生效，不可覆盖）：~/.ssh/*, ~/.aws/credentials, ~/.gnupg/*, ~/.kube/config, ~/.openharness/credentials.json 等
2. 工具显式拒绝/允许列表（denied_tools / allowed_tools）
3. 路径级别规则（fnmatch 匹配，path_rules）
4. 命令黑名单（denied_commands）
5. 按权限模式决定（FULL_AUTO 全允许，PLAN 阻写，DEFAULT 需确认）

配置示例：

{ "permission": { "mode": "default", "path_rules": [{"pattern": "/etc/*", "allow": false}], "denied_commands": ["rm -rf /", "DROP TABLE *"] } } 

生命周期钩子允许在工具执行前后插入自定义逻辑。

支持的事件：

钩子类型：

HookExecutor（hooks/executor.py）核心逻辑：

• 按 matcher 字段匹配工具名（fnmatch 语法）
• Command 钩子支持 $ARGUMENTS 模板变量注入（shell-escaped 防注入）
• Prompt/Agent 钩子要求 LLM 返回 {"ok": true/false} JSON
• 钩子可设置 block_on_failure，失败时阻止工具执行

多后端支持：

统一协议：

class SupportsStreamingMessages(Protocol): async def stream_message(self, request: ApiMessageRequest) -> AsyncIterator[ApiStreamEvent]: ... 

所有客户端实现此协议，可在运行时热替换。

Provider Registry（api/registry.py，424 行）维护了完整的 LLM 提供商元数据，支持自动检测：

• 按关键词匹配模型名
• 按 API Key 前缀识别网关（如 OpenRouter: sk-or-）
• 按 base_url 关键词识别提供商
• 分类标记：is_gateway / is_local / is_oauth

重试机制（api/client.py）：

• 最多 3 次重试
• 可重试状态码：429, 500, 502, 503, 529
• 指数退避 + 抖动（base 1s, max 30s）
• 支持 Retry-After 响应头

AuthManager（auth/manager.py）管理多 Provider 认证：

• ProviderProfile：命名的工作流配置（provider, api_format, auth_source, model, base_url 等）
• 凭据存储：~/.openharness/credentials.json，支持 per-profile 凭据槽位
• 外部凭据绑定：Claude CLI 订阅（~/.claude/.credentials.json）、Codex 订阅（~/.codex/auth.json）
• OAuth 流程：GitHub Copilot 设备码流程

配置优先级：CLI 参数 → 环境变量 → 配置文件 → 默认值

内置 Provider Profile：

Settings 模型（config/settings.py，808 行）：

class Settings(BaseModel): model: str = "sonnet" permission: PermissionSettings memory: MemorySettings sandbox: SandboxSettings mcp_servers: dict[str, McpServerConfig] hooks: dict[str, HookDefinition] theme: str = "default" # ... 更多字段 

配置解析：

• load_settings() 从 ~/.openharness/settings.json 加载
• save_settings() 持久化到磁盘
• merge_cli_overrides() 合并运行时 CLI 覆盖（不会写入磁盘）
• Model 别名解析：sonnet → claude-sonnet-4-6，opus → claude-opus-4-6

路径管理（config/paths.py）：

• 配置目录：~/.openharness/
• 数据目录：~/.openharness/data/
• 日志目录：~/.openharness/logs/
• 支持跨平台（Windows/macOS/Linux）

54 条内置斜杠命令（commands/registry.py，1605 行）：

命令上下文：

@dataclass class CommandContext: engine: QueryEngine hooks_summary: str mcp_summary: str plugin_summary: str cwd: str tool_registry: ToolRegistry | None app_state: AppStateStore | None session_backend: SessionBackend 

命令结果：

@dataclass class CommandResult: message: str | None = None should_exit: bool = False clear_screen: bool = False refresh_runtime: bool = False # 刷新运行时 submit_prompt: str | None = None # 自动提交 prompt submit_model: str | None = None # 切换模型 

McpClientManager（mcp/client.py）管理 MCP 服务器连接：

• 支持 Stdio 传输（本地进程）
• 支持 HTTP 传输（远程服务）
• 自动重连（reconnect_all()）
• 工具和资源发现（list_tools() / list_resources()）
• 调用 MCP 工具（call_tool()）
• 读取 MCP 资源（read_resource()）

MCP 工具动态注册：连接后 MCP 工具自动注册到 ToolRegistry，LLM 可直接调用。

MemoryManager（memory/manager.py）：

• 基于 Markdown 文件的持久记忆
• 支持添加/删除/搜索记忆条目
• YAML frontmatter 元数据（name, description, type）
• 搜索时同时匹配元数据和正文（元数据权重更高）
• 支持多语言分词（包括中文字符）

记忆目录：~/.openharness/memory/

自动压缩（services/compact/__init__.py，56KB — 项目最大文件）：

• Microcompact：清除旧工具结果内容（低成本）
• LLM-based 摘要：对较早消息进行完整摘要（高成本）
• Carry-over 保护：压缩后通过 tool_metadata 恢复关键状态：
  • 用户目标（task_focus_state.goal）
  • 活跃文件（active_artifacts）
  • 已验证工作（verified_state）
  • 工作日志（recent_work_log）
  • 已加载技能（invoked_skills）
  • 读取文件状态（read_file_state）
  • 异步 Agent 状态（async_agent_state）

Coordinator（coordinator/）：

• TeamRegistry：管理 Agent 团队和成员
• WorkerConfig：子 Agent 配置
• TaskNotification：任务完成通知
• Coordinator 上下文注入

Swarm（swarm/，11 个文件）：

后端类型：

• IN_PROCESS：进程内执行（共享内存）
• SUBPROCESS：子进程执行（隔离，通过 stdin/stdout 通信）

TaskManager（tasks/manager.py）管理后台任务：

• 创建、查询、更新、停止任务
• 两种任务类型：LocalAgentTask（Agent 任务）、LocalShellTask（Shell 任务）
• 支持任务输出获取

12 种渠道实现：

消息总线（channels/bus/）：

• MessageBus：进程内消息队列
• OutboundMessage：出站消息事件

渠道管理器（channels/impl/manager.py）：统一管理所有渠道适配器的生命周期。

---

ohmo 是 OpenHarness 之上的个人 Agent 应用，通过消息渠道提供长期运行的助手服务。

~/.ohmo/ ├── soul.md # Agent 人格和行为定义 ├── identity.md # Agent 身份 ├── user.md # 用户画像和偏好 ├── BOOTSTRAP.md # 首次运行引导 ├── gateway.json # Provider profile 和渠道配置 ├── memory/ # 个人记忆 ├── skills/ # 自定义技能 ├── plugins/ # 自定义插件 ├── sessions/ # 会话历史 └── logs/ # 日志 

OhmoGatewayService（ohmo/gateway/service.py）：

1. 加载 gateway.json 配置
2. 创建 MessageBus 和 OhmoSessionRuntimePool
3. 启动 OhmoGatewayBridge（消息桥接）
4. 启动 ChannelManager（渠道管理器）
5. 监听各渠道消息 → 分发给运行时池中的 Agent 会话

OhmoSessionRuntimePool（ohmo/gateway/runtime.py，28.7KB）：

• 管理多个并发的 Agent 会话
• 每个会话独立的 RuntimeBundle
• 支持会话超时和自动清理

• React TUI：交互式终端界面（launch_ohmo_react_tui()）
• Backend Only：仅运行后端服务（run_ohmo_backend()）
• Print Mode：单次查询输出（run_ohmo_print_mode()）
• Gateway：前台/后台网关服务

---

BackendHost（ui/backend_host.py）通过 JSON 协议与 React 前端通信：

• 前端通过 stdin 发送 JSON 请求
• 后端通过 stdout 返回 JSON 事件
• 使用 OPENHARNESS_FRONTEND_CONFIG 环境变量传递配置

核心 Hook：useBackendSession（hooks/useBackendSession.ts，12KB）管理整个会话状态。

内置主题：default, dark, minimal, cyberpunk, solarized

通过 ThemeProvider + ThemeContext 管理，支持运行时切换。

---

RuntimeBundle（ui/runtime.py）是运行时核心数据结构：

@dataclass class RuntimeBundle:

api_client: SupportsStreamingMessages cwd: str mcp_manager: McpClientManager tool_registry: ToolRegistry app_state: AppStateStore hook_executor: HookExecutor engine: QueryEngine commands: object external_api_client: bool enforce_max_turns: bool session_id: str settings_overrides: dict session_backend: SessionBackend extra_skill_dirs: tuple extra_plugin_roots: tuple 

组装流程（build_runtime()）：

1. 加载 Settings
2. 解析 Auth → 构建 API Client
3. 创建 MCP Manager → 连接 MCP 服务器 → 注册 MCP 工具
4. 创建 Tool Registry → 注册内置工具 + MCP 工具
5. 创建 Hook Registry + Hook Executor
6. 构建 System Prompt（环境信息 + CLAUDE.md + skills + memory）
7. 创建 Permission Checker
8. 创建 QueryEngine
9. 创建 Command Registry
10. 返回 RuntimeBundle

关键函数：

• build_runtime() — 构建运行时
• start_runtime() — 启动 MCP 连接等异步初始化
• handle_line() — 处理用户输入（命令或 prompt）
• close_runtime() — 关闭 MCP 连接等清理

---

用户输入 ↓ CLI (cli.py) / React TUI (App.tsx) ↓ run_repl() → launch_react_tui() ↓ build_runtime() → RuntimeBundle ↓ handle_line(bundle, user_input) ├── 斜杠命令? → CommandRegistry.handle() └── 普通 prompt → QueryEngine.submit_message() ↓ run_query(context, messages) ├── auto_compact_if_needed() ├── api_client.stream_message() ├── tool_use? → _execute_tool_call() │ ├── PreHook │ ├── Permission Check │ ├── tool.execute() │ ├── PostHook │ └── _record_tool_carryover() └── 返回 StreamEvent 流 ↓ UI 渲染（React TUI / rich / stdout） 

消息渠道（飞书/Slack/Telegram/Discord） ↓ ChannelAdapter → MessageBus ↓ OhmoGatewayBridge ↓ OhmoSessionRuntimePool → get_or_create_session() ↓ RuntimeBundle → QueryEngine.submit_message() ↓ Agent 响应 → OhmoGatewayBridge → ChannelAdapter ↓ 回复到消息渠道 

---

from pydantic import BaseModel, Field from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult class MyToolInput(BaseModel): query: str = Field(description="Search query") class MyTool(BaseTool): name = "my_tool" description = "Does something useful" input_model = MyToolInput async def execute(self, arguments: MyToolInput, context: ToolExecutionContext) -> ToolResult: return ToolResult(output=f"Result for: {arguments.query}") 

创建 ~/.openharness/skills/my-skill/SKILL.md：

— name: my-skill description: Expert guidance for my specific domain —

# My Skill

When to use Use when the user asks about [your domain].

Workflow 1. Step one 2. Step two

创建 .openharness/plugins/my-plugin/.claude-plugin/plugin.json：

{ “name”: “my-plugin”, “version”: “1.0.0”, “description”: “My custom plugin” } 

在子目录中添加 commands (commands/.md)、hooks (hooks/hooks.json)、agents (agents/.md)。

---

---