> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/04_runtime/4.7_miniharness_runtime.md).

# 4.7 实战：MiniHarness 运行时实现

本节使用 Python 实现一个完整、可运行的 MiniHarness 运行时引擎。涵盖智能体循环、消息系统、事件处理、工具执行和状态管理等核心概念。

完整代码参见 lab/mini\_harness/runtime/。

## 4.7.1 设计决策：消息和状态模型

“消息”是 MiniHarness 的核心抽象。系统使用分块(block)结构组织内容，允许混合文本和工具调用。

关键设计点：

* **TextBlock**：“纯文本内容，支持多块拼接”
* **ToolUseBlock**：“智能体要求执行的工具调用，包含输入参数”
* **ToolResultBlock**：“工具执行结果，包括成功/失败状态”
* **Message**：“消息容器，role 标识发送者，content 是块数组”
* **AgentState**：“会话状态跟踪，维护消息历史和轮次计数”

参考实现片段（models.py；实际代码中该类名为 `RuntimeMessage`，书中为行文简洁记作 `Message`）：

```python
@dataclass
class Message:
    role: str
    content: List[Any]

    @classmethod
    def user(cls, text: str) -> "Message":
        return cls(role="user", content=[TextBlock(text=text)])

    def has_tool_calls(self) -> bool:
        return any(isinstance(block, ToolUseBlock)
                  for block in self.content)

    def get_tool_calls(self) -> List[ToolUseBlock]:
        return [b for b in self.content
                if isinstance(b, ToolUseBlock)]
```

## 4.7.2 事件驱动架构

系统通过“异步事件流”向外部通知执行阶段。每个重要事件都包含 metadata，便于监控和日志记录。

事件类型：

* AGENT\_START, AGENT\_END：“会话生命周期”
* TURN\_START, TURN\_END：“每个智能体循环轮次”
* TEXT\_RESPONSE:“智能体生成的文本内容”
* TOOL\_EXECUTE, TOOL\_RESULT:“工具调用和结果”
* ERROR:“执行异常”

参考片段(events.py)：

```python
class EventType(Enum):
    AGENT_START = "agent_start"
    TURN_START = "turn_start"
    TOOL_EXECUTE = "tool_execute"
    # ... 更多类型

@dataclass
class Event:
    event_type: EventType
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    metadata: Dict[str, Any] = field(default_factory=dict)

    def to_json(self) -> str:
        return json.dumps({
            "event_type": self.event_type.value,
            "timestamp": self.timestamp.isoformat(),
            "metadata": self.metadata
        })
```

## 4.7.3 智能体循环核心逻辑

运行时引擎(RuntimeEngine)实现了推理、工具执行、反馈融合的完整循环。

循环流程：

1. **初始化**：创建会话，输入用户提示
2. **推理轮次**（最多 max\_turns）
   * 调用 \_infer() 获取智能体响应
   * 如果包含工具调用，执行并收集结果
   * 将结果添加到状态，继续下一轮
   * 如果无工具调用，循环结束
3. **关闭**：生成 AGENT\_END 事件

关键设计：” **Token 预算管理** “（简化实现中为常量 4000），实际系统需计算累积 Token 数并在超出时截断历史。

参考片段(engine.py)：

```python
async def run(self, user_input: str) -> AsyncIterator[Event]:
    yield AgentStartEvent(metadata={"user_input": user_input})
    session_id = f"sess_{uuid.uuid4().hex[:8]}"
    state = AgentState(session_id=session_id)
    state.add_message(Message.user(user_input))

    for turn in range(self.max_turns):
        yield TurnStartEvent(metadata={"turn_number": turn})
        response = await self._infer(state)

        if response is None:
            yield ErrorEvent(metadata={"error": "Inference failed"})
            break

        state.add_message(response)
        tool_calls = response.get_tool_calls()

        if tool_calls:
            for tool_use in tool_calls:
                result = await self._execute_tool(tool_use)
                state.add_message(Message.tool_result(result))
        else:
            break

        yield TurnEndEvent(metadata={"turn_number": turn})

    yield AgentEndEvent(metadata={"session_id": session_id})
```

## 4.7.4 工具执行和错误处理

\_execute\_tool() 方法从注册表中查找工具并调用。所有异常都被捕获并转换为 ToolResultBlock，确保循环不中断。

设计原则：“ **故障隔离** ”——单个工具失败不应导致整个会话失败。

```python
async def _execute_tool(self, tool_use: ToolUseBlock) -> ToolResultBlock:
    tool = self.tool_registry.get(tool_use.name)
    if not tool:
        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=f"Tool not found",
            is_error=True,
            error_type="ToolNotFoundError"
        )

    try:
        result = await tool.call(tool_use.input)
        # Tool.call() 返回 ToolResult 数据类（success/content/execution_time/error_type），
        # 需要解包成 ToolResultBlock 期望的字符串 content + is_error 字段。
        if hasattr(result, "success") and hasattr(result, "content"):
            return ToolResultBlock(
                tool_use_id=tool_use.id,
                content=str(result.content),
                is_error=not result.success,
                error_type=getattr(result, "error_type", None),
            )
        # 向后兼容：若工具直接返回字符串，按成功处理
        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=str(result),
            is_error=False
        )
    except Exception as e:
        return ToolResultBlock(
            tool_use_id=tool_use.id,
            content=str(e),
            is_error=True,
            error_type=type(e).__name__
        )
```

## 4.7.5 应用组合入口与工具调用恢复

`RuntimeEngine` 保留为便于阅读的最小事件循环；可运行示例 `lab/examples/simple_agent.py` 则使用 `lab/mini_harness/application.py` 的 `HarnessApplication` 作为组合入口。这个入口把原先容易散落在调用方的能力固定为同一顺序：上下文组装 → 权限/护栏 → 检查点 → 可恢复错误重试 → 本地或 MCP 工具 → 完成事件。

```python
from mini_harness.application import HarnessApplication
from mini_harness.runtime.checkpoint import JSONCheckpointStore

application = HarnessApplication(
    tool_registry=registry,
    mcp_registry=mcp_registry,
    context_assembler=context_assembler,
    secure_executor=secure_executor,
    checkpoint_store=JSONCheckpointStore("runtime-checkpoints.json"),
)

result = await application.execute_tool(
    "file_read",
    {"path": "README.md"},
    user_id="user-1",
    session_id="session-1",
    call_id="tool-call-1",
)
```

检查点以 `session_id + call_id` 标识一次调用，显式保存 `user_id` 所代表的 principal；参数指纹也包含该 principal、工具名和参数。每次调用都先重新经过权限与护栏，再原子认领检查点，因此策略已改为 `DENY` 的请求无法读取旧结果。只有获得随机 Lease 的 Owner 能执行和完成调用，其他并发请求只能回放已完成结果，或对 `started` 状态明确拒绝重放。恢复时：

* `completed`：仅向同一 principal 返回已经保存的结果，不再次调用工具；
* `started`：拒绝自动重放；这既可能表示当前 Owner 正在执行，也可能表示进程在副作用完成后、结果落盘前崩溃；
* 同一调用 ID 被其他 principal 使用，或参数指纹不同：抛出冲突错误。

JSON 存储对同一绝对路径的本进程实例共享线程锁，并使用同目录临时文件、`fsync` 和原子替换，把文件权限设置为仅当前用户可读写。它解决任务/线程并发和本机进程重启后的重复调度问题，但不是跨进程协调器，也不能让远端副作用与本地文件形成分布式事务；多进程部署应改用带条件写入或事务的存储，写操作仍应向服务端传递幂等键。

## 4.7.6 主要特性总结

1. **消息块架构**：“灵活组合文本和工具”，支持混合内容
2. **异步事件流**：“非阻塞监控”，易于集成日志、指标收集
3. **会话状态隔离**：“独立的 AgentState”，支持并发运行多个会话
4. **工具注册表**：“动态工具加载”，可在运行时增删工具
5. **故障恢复**：“优雅降级”，工具失败不影响会话继续

## 4.7.7 扩展方向

生产系统需补充：

* **真实 LLM 集成**：“替换 \_infer() 的模拟实现”
* **上下文窗口管理**：“动态截断旧消息以保持在 Token 预算内”
* **会话级持久化**：当前已实现工具调用检查点；完整消息历史、模型轮次和 AgentState 恢复仍需持久化
* **丰富的工具库**：“文件操作、网络、数据库等”
* **高级特性**：“意图分类、工具选择器优化、多智能体协调”

## 4.7.8 本节小结

MiniHarness 展示了生产级智能体运行时的关键架构模式：

1. **分层设计**：“模型层、事件层、工具层、引擎层”
2. **异步驱动**：“Python asyncio 实现高效并发”
3. **事件中心**：“所有状态变化都生成可观测的事件”
4. **容错设计**：“隔离失败，保证系统稳定”

这些原则直接应用于 Claude Code 和 OpenClaw 等生产系统。参考 lab/mini\_harness/runtime/ 查看完整实现代码。