附录 D：MiniHarness 实战项目

MiniHarness 是本书配套的实战项目——一个最小但完整的 Agent Harness 系统，使用 Python 实现。源代码位于 lab/ 目录。

项目概览

MiniHarness 从架构、工具调用、记忆、编排、安全、评估等方面全方位展示 Harness 框架的核心设计原理。通过本项目，读者可以理解 Harness 系统的完整架构，学习安全防护的具体实现，掌握评估框架的搭建，并在此基础上构建生产级系统。

项目特性包括：完整的工具调用框架、多层安全防护（权限、路径校验、护栏）、评估测试体系、可观测性与日志基础设施。

工作原理

examples/simple_agent.py 实现了一个约 340 行的完整 Agent，展示了 Harness 的核心循环：

关键组件对应关系：

示例中的代码	MiniHarness 模块	书中章节
LLMClient	models/provider.py → OpenAIProvider	第7章
ToolRegistry + FileReadTool	tools/registry.py + tools/builtin.py	第5章
HarnessApplication	application.py	第4、9、11、12章的集成边界
工具调用检查点	runtime/checkpoint.py	第11章
SimpleAgent.run() 循环	runtime/engine.py → RuntimeEngine	第4章
流式事件输出	runtime/events.py	第4章

目录结构

MiniHarness项目的完整目录结构：

核心模块代码索引

0. 应用组合入口

mini_harness/application.py

关键类：HarnessApplication

主要方法：

• prepare_context()：组装记忆上下文，事件只记录长度而不记录正文

• execute_tool()：让本地与 MCP 工具共用权限、护栏、重试、事件和检查点

• list_tools()：向模型适配层提供本地工具 Schema

1. 消息和事件

mini_harness/core/message.py、mini_harness/core/event.py

关键类：Message、Event

主要方法：

• 消息格式定义

• 事件类型枚举

代码位置：第2章详细代码实现

2. Tool基类和智能体定义

mini_harness/core/tool.py、mini_harness/core/agent.py

关键类：Tool、Agent

主要方法：

• 工具定义接口

• Agent状态管理

代码位置：第2章完整实现

3. 运行时引擎

mini_harness/runtime/engine.py、mini_harness/runtime/checkpoint.py

关键类：RuntimeEngine

主要方法：

• run(): 主智能体循环，生成运行时事件流

• _infer(): 模拟模型推理

• _execute_tool(): 执行工具调用并转换为 ToolResultBlock

• JSONCheckpointStore: 原子写入工具调用状态和可重放结果

代码位置：第4章详细代码实现

4. 工具注册表

mini_harness/tools/registry.py

关键类：ToolRegistry

主要方法：

• register(): 注册工具

• get(): 获取工具

• list_tools(): 列出所有工具

代码位置：第5章完整实现

5. 内置工具

mini_harness/tools/builtin.py

关键类：内置工具实现

主要方法：

• 标准工具的完整实现

代码位置：第5章详细代码

6. 记忆存储

mini_harness/memory/storage.py、mini_harness/memory/context.py、mini_harness/memory/consolidation.py

关键类：MemoryStore、MemoryEntry、ContextAssembler、ConsolidationEngine

主要方法：

• 长期记忆存储

• 上下文管理

• 记忆整合

代码位置：第6章完整实现

7. 模型提供者

mini_harness/models/provider.py、mini_harness/models/parser.py、mini_harness/models/quality.py

关键类：BaseProvider、OpenAIProvider、ClaudeProvider、ResponseParser、QualityGate

主要方法：

• 模型调用接口

• 输出解析逻辑

• 质量评估

代码位置：第7章详细代码实现

8. 编排引擎

mini_harness/orchestration/engine.py

关键类：OrchestrationEngine

主要方法：

• 复杂工作流编排

• 多智能体协调

代码位置：第8章完整实现

9. MCP集成

mini_harness/mcp/client.py、mini_harness/mcp/transports.py、mini_harness/mcp/auth.py、mini_harness/mcp/integration.py

关键类：MCPClient、MCPToolRegistry、MCPToolAdapter、MiniHarnessWithMCP

主要方法：

• 官方 SDK 的 stdio 与 Streamable HTTP 生命周期

• Bearer Token、自定义请求头、超时取消和显式关闭

• 工具发现、Schema 缓存与 LLM 格式适配

代码位置：第9章详细代码

10. 生产化加固

mini_harness/utils/config.py

关键类：Config

主要方法：

• 配置管理

• 生产参数调优

代码位置：第10章完整实现

11. 可观测性和可靠性

mini_harness/reliability/tracing.py、mini_harness/reliability/monitoring.py、mini_harness/reliability/logging.py、mini_harness/reliability/resilience.py

关键类：Span、TraceCollector、MonitoringSystem、StructuredLogger、RetryDecorator、CircuitBreaker

主要方法：

• 链路追踪

• 指标收集

• 日志记录

• 可靠性保障

代码位置：第11章详细代码实现

12. 权限系统

mini_harness/security/permissions.py

关键类：

• PermissionDecisionEngine: 权限决策核心

• PermissionPolicy: 工具权限策略

主要方法：

• decide(): 做出权限决策(ASK/AUTO/DENY)

• record_approval(): 记录用户批准

• get_audit_logs(): 读取权限决策审计记录

代码位置：第12.2节详细代码

13. 路径校验

mini_harness/security/path_validator.py

关键类：PathValidator

主要方法：

• validate(): 5层路径校验

  • _check_length(): 第1层 - 长度检查

  • _decode_all_encodings(): 第2层 - URL解码

  • _normalize_unicode(): 第3层 - Unicode规范化

  • _normalize_platform(): 第4层 - 平台规范化

  • _resolve_and_check_boundaries(): 第5层 - realpath + 边界检查

代码位置：第12.4节完整实现

14. 护栏框架

mini_harness/security/guardrails.py

关键类：

• DangerousCommandDetector: 危险命令检测

主要方法：

• detect(): 检测危险命令

• get_reason(): 返回危险命令命中的原因

代码位置：第12.3节完整实现

15. 评估系统

lab/tests/

关键类：测试套件

主要方法：

• 单元测试

• 集成测试

• 安全测试

• 端到端和性能测试属于扩展目标，当前仓库未提供独立测试层

代码位置：第13章完整实现

快速开始

安装与配置

# 克隆仓库
git clone https://github.com/yeasy/harness_engineering_guide.git
cd harness_engineering_guide/lab

# 创建虚拟环境并安装
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -e ".[dev]"

MiniHarness 兼容所有 OpenAI API 格式的 LLM 服务。复制 .env.example 并配置：

cp .env.example .env

# 支持的服务（任选其一）
# OpenAI
read -rsp "LLM_API_KEY: " LLM_API_KEY; echo; export LLM_API_KEY
export LLM_BASE_URL="https://api.openai.com/v1" LLM_MODEL="gpt-5.4-mini"
# DeepSeek
export LLM_API_KEY="<LLM_API_KEY>" LLM_BASE_URL="https://api.deepseek.com" LLM_MODEL="deepseek-chat"
# Ollama 本地模型(无需付费 API Key)
export LLM_API_KEY="ollama" LLM_BASE_URL="http://localhost:11434/v1" LLM_MODEL="qwen2.5:7b"

运行示例

# 命令行传入任务
python examples/simple_agent.py "读取 README.md 并总结项目"

# 交互式输入
python examples/simple_agent.py

运行测试

# 全部测试
pytest tests/ -v

# 只跑某个模块
pytest tests/unit/test_security.py -v
pytest tests/integration/ -v

# 覆盖率报告
pytest tests/ --cov=mini_harness --cov-report=html

使用 MiniHarness 库

除了运行示例，还可以在自己的代码中导入 MiniHarness 模块：

import asyncio
from mini_harness.tools.builtin import BashTool, FileReadTool, FileWriteTool
from mini_harness.tools.registry import ToolRegistry
from mini_harness.models.provider import (
    ModelConfig, ModelProviderType, create_provider, ProviderMessage
)

# 1. 注册工具
registry = ToolRegistry()
registry.register(BashTool())
registry.register(FileReadTool())
registry.register(FileWriteTool())

# 2. 创建 LLM Provider
config = ModelConfig(
    provider=ModelProviderType.OPENAI,
    model_id="deepseek-chat",
    api_key="<LLM_API_KEY>",
    base_url="https://api.deepseek.com",
)
provider = create_provider(config)

# 3. 调用 LLM（带工具）
tools = registry.list_tools()
response = provider.complete_with_tools(
    messages=[ProviderMessage("user", "用 bash 查看系统信息")],
    tools=tools,
)
print(response.content)
print(response.tool_calls)

使用熔断器做故障转移

from mini_harness.models.provider import ModelConfig, ModelProviderType, ModelSelectionEngine, ProviderMessage

primary = ModelConfig(ModelProviderType.OPENAI, "gpt-5.4", api_key="<LLM_API_KEY>")
fallback = ModelConfig(ModelProviderType.OPENAI, "gpt-5.4-mini", api_key="<LLM_API_KEY>")

engine = ModelSelectionEngine(primary, fallback_chain=[fallback])
provider = engine.select_model()  # 自动选择可用的模型

try:
    response = provider.complete([ProviderMessage("user", "Hello")])
    engine.mark_success(provider.config.model_id)
except Exception:
    engine.mark_failure(provider.config.model_id)
    # 下次调用 select_model() 会自动切换到 fallback

架构总览图

MiniHarness的整体架构由多个层级组成，以下是完整的系统架构关系：

文件到章节的映射

文件	对应章节	关键概念
core/message.py	2	消息类型定义
core/tool.py	2	Tool基类定义
core/agent.py	2	Agent定义
core/event.py	2	事件系统
runtime/engine.py	4	执行流程和循环
runtime/models.py	4	模型管理
runtime/events.py	4	运行时事件
tools/registry.py	5	工具注册和管理
tools/builtin.py	5	内置工具实现（BashTool、FileReadTool、FileWriteTool、ExecutionPipeline）
memory/storage.py	6	记忆存储
memory/context.py	6	上下文管理
memory/consolidation.py	6	记忆整合
models/provider.py	7	模型提供者
models/parser.py	7	输出解析
models/quality.py	7	质量评估
orchestration/engine.py	8	编排引擎
mcp/integration.py	9	MCP集成
utils/config.py	10	生产化加固
reliability/tracing.py	11	链路追踪
reliability/monitoring.py	11	指标收集
reliability/logging.py	11	日志系统
security/permissions.py	12.2	权限系统
security/path_validator.py	12.4	路径校验
security/guardrails.py	12.3	护栏防护
tests/	13	测试框架

扩展和集成点

添加新工具

示例如下：

# 在 tools/ 目录创建 new_tool.py
from mini_harness.core.tool import Tool
from mini_harness.core.tool import ToolResult

class MyCustomTool(Tool):
    def name(self) -> str:
        return "my_tool"

    def description(self) -> str:
        return "描述"

    def input_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {"text": {"type": "string"}},
            "required": ["text"],
        }

    async def call(self, params: dict) -> ToolResult:
        # 实现逻辑
        return ToolResult(success=True, content=params["text"], execution_time=0.0)

# 通过工具注册表注册
from mini_harness.tools.registry import ToolRegistry
registry = ToolRegistry()
registry.register(MyCustomTool())

自定义测试

代码如下：

# 在 tests/ 目录创建 test_custom.py
from mini_harness.runtime import RuntimeEngine
from mini_harness.tools.registry import ToolRegistry
import pytest

@pytest.mark.asyncio
async def test_my_feature():
    registry = ToolRegistry()
    engine = RuntimeEngine(tool_registry=registry)
    # 自定义测试逻辑
    async for event in engine.run("test input"):
        pass

集成新的大语言模型

示例如下：

# 在 models/ 目录创建 new_llm_provider.py
from mini_harness.models.provider import BaseProvider, ModelConfig, ModelProviderType, ProviderResponse
from typing import List, Dict, Optional, Generator

class MyLLMProvider(BaseProvider):
    def complete(self, messages: List, tools: Optional[List[Dict]] = None) -> ProviderResponse:
        # 调用新的LLM
        return ProviderResponse(content="Response", tokens_used=0, model="custom-llm")

    def stream(self, messages: List, tools: Optional[List[Dict]] = None) -> Generator[str, None, None]:
        yield "Streamed response"

    def complete_with_tools(self, messages: List, tools: List[Dict]) -> ProviderResponse:
        return self.complete(messages, tools=tools)

性能基准

在标准硬件上的参考指标（仅供参考）：

操作	延迟	吞吐量
工具调用	10-50ms	100-200 calls/sec
路径校验	<1ms （缓存）	10000+ validations/sec
权限决策	5-10ms	1000-2000 decisions/sec
测试执行	<100ms	可实时运行

---

仓库地址：https://github.com/yeasy/harness_engineering_guide

获取最新版本：参见本书各章节的 MiniHarness 实战部分

问题反馈：欢迎在GitHub Issues中反馈bug和建议