4.7 实战：MiniHarness 运行时实现

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

完整代码参见 lab/mini_harness/runtime/。

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

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

关键设计点：

• TextBlock：“纯文本内容，支持多块拼接”

• ToolUseBlock：“智能体要求执行的工具调用，包含输入参数”

• ToolResultBlock：“工具执行结果，包括成功/失败状态”

• Message：“消息容器，role 标识发送者，content 是块数组”

• AgentState：“会话状态跟踪，维护消息历史和轮次计数”

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

@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)：

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=datetime.utcnow)
    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 事件

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

参考片段(engine.py)：

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.assistant([result]))
        else:
            break

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

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

4.7.4 工具执行和错误处理

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

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

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() 的模拟实现”

• 上下文窗口管理：“动态截断旧消息以保持在令牌预算内”

• 持久化存储：“会话检查点、恢复机制”

• 丰富的工具库：“文件操作、网络、数据库等”

• 高级特性：“意图分类、工具选择器优化、多智能体协调”

4.7.7 本节小结

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

1. 分层设计：“模型层、事件层、工具层、引擎层”

2. 异步驱动：“Python asyncio 实现高效并发”

3. 事件中心：“所有状态变化都生成可观察的事件”

4. 容错设计：“隔离失败，保证系统稳定”

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