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）：

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

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，确保循环不中断。

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

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 工具 → 完成事件。

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/ 查看完整实现代码。