d2b4c5029d
新增 mcfold/events.py：EventLog（环形缓冲 + seq + 游标）
- 不做回调式推送：LLM 是回合制的，回调会让事件在两次观察之间静默丢失
- drain(limit) 只取本轮、剩余留待下次（不丢）
- 细分层（block_changed❌y:z）独立可查，不污染粗粒度流
- render_events() 输出紧凑文本行，省 token
session 接线：block_change / multi_block_change / chat / damage /
entity spawn·move·destroy / held_item_slot，并加 chat()/command() 发送
修的真 bug：
- dropped 恒 0：deque(maxlen=) 丢元素无声 ⇒ 改手动计数（事件丢了必须说）
- 事件键 to_state_id 与渲染器的 to 不一致 ⇒ 显示 None，改为解出方块名；
from 区分「未加载/界外」而非裸 None
- run_tests.sh 只数 ✓ 不数 ★ ⇒ 判据少报（98 修正为 110）
实测：/setblock 后事件正确报 grass_block→stone→air，与状态层一致
判据：110 项全通过；4 项变异全部被抓
213 lines
9.0 KiB
Python
213 lines
9.0 KiB
Python
"""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)
|