"""M4 事件层：把状态变化变成 **LLM 可按需消费** 的事件流。 ## 为什么不是回调，而是「环形缓冲 + 游标」 mineflayer 的 `bot.on('physicsTick', …)` 是**回调**式，适合长跑的机器人。 但我们要服务的是 **LLM agent**： - LLM 是**回合制**的：一次思考 → 一次工具调用 → 一次观察。它不会一直在跑。 - 回调式对 LLM 有三个坏处： 1. 事件在两次观察之间发生 ⇒ **静默丢失**，agent 以为世界没变； 2. 要在回调里"攒"上下文 ⇒ 状态散落在闭包里，不可查询、不可重放； 3. 事件粒度不可控（physicsTick 每秒 20 次 ⇒ 直接把 token 烧穿）。 ⇒ 所以事件层是**可排空的环形缓冲**： agent 每次观察时调 `drain()`，把「上次观察之后发生的事」按需取走； 每条带**序号**（`seq`），可精确追问「我看到 seq=37 之后又发生了什么」。 ## 分级：粗粒度（默认给 agent） vs 细分层（按需） - **粗粒度**：`block_changed`、`entity_moved`、`chat`、`damage`、`joined`。 这些是「游戏里发生了什么」级别的变化，一轮观察给个 20 条就够判断。 - **细分层**：`block_changed:9:4:12`（某坐标方块变了）、`chat:Alice`。 只在 agent 明确追问时用 `drain(fine=...)` 取 —— 常规轮次**不烧 token**。 ## 为什么这层不做「决策」 事件层回答「**变了什么**」，不回答「**该怎么办**」。 `chat` 事件告诉 agent 有人说了什么；该不该回、回什么，是 agent 的事。 同样，`block_changed` 说方块变了；该不该挖，agent 判断。 """ from __future__ import annotations from collections import deque from typing import Any, Iterable # ---------- 事件构造 ---------- def _ev(seq: int, kind: str, data: dict[str, Any]) -> dict[str, Any]: """统一事件形状。`kind` 粗粒度，`fine` 细分层（可选）。""" return {"seq": seq, "kind": kind, "data": data} # ---------- 缓冲 ---------- class EventLog: """环形事件缓冲 + 单调游标。 ## 三个刻意的不对称 1. **计数不限报，实现有界**：deque 满了丢最旧的，但 `dropped` 计数如实报 —— 静默丢事件会让 agent 以为世界没发生事。 2. **游标是调用方的**：`drain()` 返回后自己推进游标； 同一游标重复调只拿新事件（幂等，不会重复读）。 3. **fine 事件与粗粒度并存**：同一次变化写两条（`block_changed` 与 `block_changed:9:4:12`），但 `drain()` 默认**只给粗粒度**。 """ #: 缓冲上限。超了丢最旧并计 dropped。 CAPACITY = 4096 def __init__(self, capacity: int = CAPACITY): # ★ 不用 deque(maxlen=) ：它丢元素是**完全无声**的。 # 实测：deque(maxlen=4) 塞 10 个 ⇒ 只剩 [6,7,8,9]，没有任何痕迹。 # 而“事件被丢了却不说”正是本层最不能犯的错 —— agent 会以为 # 世界没发生变化。改用手动判满 + 显式计数。 self._cap = capacity self._buf: deque[dict[str, Any]] = deque() self._seq = 0 self._cursor = 0 # 调用方已消费到哪 self.dropped = 0 # 因容量不足丢掉的条数 self._fine: deque[dict[str, Any]] = deque() # ---------- 写入 ---------- def emit(self, kind: str, data: dict[str, Any] | None = None) -> dict[str, Any]: """记一个**粗粒度**事件。满了就丢最旧，并**显式计数**。""" self._seq += 1 e = _ev(self._seq, kind, data or {}) if len(self._buf) >= self._cap: self._buf.popleft() self.dropped += 1 self._buf.append(e) return e def emit_fine(self, fine: str, data: dict[str, Any] | None = None) -> dict[str, Any]: """记一个**细分层**事件（如 `block_changed:9:4:12`）。不进粗粒度流。""" self._seq += 1 e = _ev(self._seq, fine, data or {}) if len(self._fine) >= self._cap: self._fine.popleft() self.dropped += 1 self._fine.append(e) return e def emit_change(self, kind: str, fine: str, data: dict[str, Any] | None = None) -> None: """一次变化同时记粗 + 细（最常见：方块/实体变化）。""" self.emit(kind, data) if fine: self.emit_fine(fine, data) # ---------- 读取 ---------- @property def seq(self) -> int: """当前最大序号（= 已产生的粗+细事件总数）。""" return self._seq @property def cursor(self) -> int: return self._cursor def peek(self) -> list[dict[str, Any]]: """看看有什么，但不消费。""" return [e for e in self._buf if e["seq"] > self._cursor] def drain(self, limit: int = 40) -> list[dict[str, Any]]: """取出游标之后的新粗粒度事件，**最多 limit 条**，并推进游标。 ★ 为什么要 limit：LLM 上下文有限。超出的留给下次（不丢）， 但要在返回里说清「还有多少没看」，否则 agent 会误以为看全了。 """ pending = [e for e in self._buf if e["seq"] > self._cursor] out = pending[:limit] if out: self._cursor = out[-1]["seq"] return out def drain_fine(self, prefix: str, limit: int = 20) -> list[dict[str, Any]]: """按细分层前缀查（如 `block_changed:`），**不推进粗粒度游标**。""" out = [e for e in self._fine if e["kind"].startswith(prefix)][:limit] return out def reset_cursor(self) -> None: """把游标拉到最新（agent 明确说「我不关心之前」时用）。""" self._cursor = self._seq def stats(self) -> dict[str, Any]: return { "seq": self._seq, "cursor": self._cursor, "coarse_pending": sum(1 for e in self._buf if e["seq"] > self._cursor), "fine_total": len(self._fine), "dropped": self.dropped, # 容量不足丢掉的 } # ---------- 事件到 M3 状态的接线 ---------- #: 状态层每类变化对应的粗粒度事件名。刻意保持少而清晰。 #: —— 事件名是给 agent 的**稳定契约**，改名字会破坏 agent 的 prompt。 COARSE_KINDS = { "player_moved", # 自己移动 "joined", # 进场 "chat", # 聊天/系统消息 "block_changed", # 方块变了 "entity_spawned", # 实体出现 "entity_moved", # 实体移动 "entity_removed", # 实体消失 "damage", # 受伤/死亡 "health_changed", # 生命/饥饿 "chunk_loaded", # 新区块 "chunk_unloaded", # 区块卸载 "held_item", # 手持物变化 "inventory", # 背包变化 } def render_events(events: Iterable[dict[str, Any]]) -> str: """把事件渲染成给 LLM 读的一行行文本（省 token 的关键）。 ★ 刻意做成「文本行」而非嵌套 JSON：LLM 读 `block_changed @ 9,4,12 air` 比读 `{"kind":"block_changed","data":{"pos":[9,4,12],"from":"air","to":"stone"}}` 省一半以上 token，且更易扫读。 """ lines: list[str] = [] for e in events: k, d = e["kind"], e["data"] if k == "block_changed": lines.append( f"#{e['seq']} 方块变化 @({d.get('x')},{d.get('y')},{d.get('z')}) " f"{d.get('from')} → {d.get('to')}") elif k == "entity_spawned": lines.append(f"#{e['seq']} 出现 {d.get('kind')} id={d.get('entity_id')}") elif k == "entity_moved": lines.append( f"#{e['seq']} {d.get('kind')} id={d.get('entity_id')} 移动到 " f"({d.get('x')},{d.get('y')},{d.get('z')}) 距 {d.get('distance')}") elif k == "entity_removed": lines.append(f"#{e['seq']} 消失 id={d.get('entity_id')}") elif k == "player_moved": lines.append( f"#{e['seq']} 我移动到 ({d.get('x')},{d.get('y')},{d.get('z')})") elif k == "chat": who, what = d.get("from"), d.get("text") lines.append(f"#{e['seq']} 聊天 {who}: {what}" if who else f"#{e['seq']} 系统: {what}") elif k == "damage": lines.append(f"#{e['seq']} 受到伤害 {d.get('amount')} 血量 {d.get('health')}") elif k == "health_changed": lines.append(f"#{e['seq']} 生命/饥饿 → {d.get('health')}/{d.get('food')}") elif k == "chunk_loaded": lines.append(f"#{e['seq']} 区块加载 ({d.get('x')},{d.get('z')})") elif k == "chunk_unloaded": lines.append(f"#{e['seq']} 区块卸载 ({d.get('x')},{d.get('z')})") elif k == "joined": lines.append(f"#{e['seq']} 进入世界 ({d.get('dimension')})") elif k == "held_item": lines.append(f"#{e['seq']} 手持 {d.get('item')}") else: # 未知事件：如实标注，不静默吞 lines.append(f"#{e['seq']} {k} {d}") return "\n".join(lines)