# 4.7 实战：MiniHarness 运行时实现

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

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

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

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

关键设计点：

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

参考实现片段(models.py)：

```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 主要特性总结

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

## 4.7.6 扩展方向

生产系统需补充：

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

## 4.7.7 本节小结

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

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

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


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/04_runtime/4.7_miniharness_runtime.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.