📜 Event Sourcing

核心思想

用一个例子来理解：

银行从不说"你的余额是 X"——银行说"你发生了这些交易"。余额是交易事件的投影（projection），不是被存储的原始数据。Event Sourcing 就是把这种思维应用到软件系统中。

事件流可视化

Event Sourcing 的三大能力

LangGraph Checkpoint = Event Sourcing

LangGraph Checkpoint 代码示例

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# 1. 定义 Agent 状态
class AgentState(TypedDict):
    messages: list[BaseMessage]
    next_action: str
    tool_results: list[dict]

# 2. 创建带 Checkpoint 的图
workflow = StateGraph(AgentState)
workflow.add_node("think", think_node)
workflow.add_node("act", act_node)
workflow.add_node("observe", observe_node)
workflow.add_edge("think", "act")
workflow.add_edge("act", "observe")
workflow.add_edge("observe", "think")

# 3. 使用 PostgresSaver 持久化 checkpoint（Event Sourcing!）
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = workflow.compile(checkpointer=checkpointer)

# 4. 运行 Agent——每个步骤都是一个"事件"
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [HumanMessage("帮我订机票")]}, config)

# 5. Time Travel——回到某个 checkpoint
history = list(app.get_state_history(config))
for state in history:
    print(f"Checkpoint: {state.config['configurable']['checkpoint_id']}")
    print(f"  Next: {state.next}")
    print(f"  Values: {state.values}")

# 6. 从某个 checkpoint 重新执行（Event Replay!）
old_checkpoint = history[3]  # 回到第4个步骤
app.invoke(None, old_checkpoint.config)  # 从这里继续执行

# 7. 分叉执行（branching）
app.invoke(
    {"messages": [HumanMessage("换一个航班")]},
    old_checkpoint.config  # 从同一个 checkpoint 分叉
)

KurrentDB：Event Sourcing 专用数据库

为什么不用 PostgreSQL？

KurrentDB 的核心论点："Stop fighting your database." 用 CRUD 数据库做 Event Sourcing，就像用锤子拧螺丝——能做，但痛苦且不可靠。

Kafka：事件流平台

Apache Kafka 是 Event Sourcing 的另一个重要工具，但定位与 KurrentDB 不同：

在实践中，两者经常配合使用：KurrentDB 做 Event Store（存储业务事件），Kafka 做 Event Bus（分发集成事件）。

Event Sourcing 实战

Python 实现（无框架）

import json
from dataclasses import dataclass
from datetime import datetime
from typing import Any

# 1. 定义事件
@dataclass
class Event:
    type: str
    data: dict
    timestamp: str
    version: int

# 2. Event Store（简化版）
class EventStore:
    def __init__(self):
        self.streams: dict[str, list[Event]] = {}
    
    def append(self, stream_id: str, events: list[Event], expected_version: int = -1):
        """追加事件到流，带乐观并发控制"""
        stream = self.streams.setdefault(stream_id, [])
        if stream and stream[-1].version != expected_version:
            raise ConcurrencyError(f"Expected version {expected_version}, got {stream[-1].version}")
        for i, event in enumerate(events):
            event.version = expected_version + 1 + i
            stream.append(event)
    
    def read(self, stream_id: str, from_version: int = 0) -> list[Event]:
        """读取流中的事件"""
        return [e for e in self.streams.get(stream_id, []) if e.version >= from_version]

# 3. 聚合根（Aggregate）
class BankAccount:
    def __init__(self, stream_id: str, store: EventStore):
        self.stream_id = stream_id
        self.store = store
        self.balance = 0
        self.is_frozen = False
        self.version = -1
        self._uncommitted: list[Event] = []
    
    def _apply(self, event: Event):
        """应用事件到状态——这是 projection"""
        if event.type == "AccountCreated":
            self.balance = 0
            self.is_frozen = False
        elif event.type == "Deposited":
            self.balance += event.data["amount"]
        elif event.type == "Withdrawn":
            self.balance -= event.data["amount"]
        elif event.type == "AccountFrozen":
            self.is_frozen = True
        elif event.type == "AccountUnfrozen":
            self.is_frozen = False
        self.version = event.version
    
    def load(self):
        """从事件流重建状态"""
        for event in self.store.read(self.stream_id):
            self._apply(event)
    
    def deposit(self, amount: int):
        """业务命令 → 产生事件"""
        if amount <= 0:
            raise ValueError("Amount must be positive")
        self._uncommitted.append(Event(
            type="Deposited",
            data={"amount": amount},
            timestamp=datetime.now().isoformat(),
            version=0  # will be set on commit
        ))
        # 立即应用到本地状态
        self._apply(self._uncommitted[-1])
    
    def withdraw(self, amount: int):
        if self.is_frozen:
            raise AccountFrozenError()
        if amount > self.balance:
            raise InsufficientFundsError()
        self._uncommitted.append(Event(
            type="Withdrawn",
            data={"amount": amount},
            timestamp=datetime.now().isoformat(),
            version=0
        ))
        self._apply(self._uncommitted[-1])
    
    def commit(self):
        """提交未持久化的事件"""
        if self._uncommitted:
            self.store.append(self.stream_id, self._uncommitted, self.version - len(self._uncommitted))
            self._uncommitted = []

# 4. 使用
store = EventStore()
account = BankAccount("acc-001", store)
account.load()          # 重建状态
account.deposit(100)    # 余额: 100
account.withdraw(20)    # 余额: 80
account.commit()        # 持久化事件

# 5. 时间旅行——回到第2个事件
account2 = BankAccount("acc-001", store)
for event in store.read("acc-001")[:2]:  # 只重放前2个事件
    account2._apply(event)
print(f"余额（事件2时）: {account2.balance}")  # 100

🤖 Agent 系统中的 Event Sourcing

// Agent 事件流示例
[
  { type: "user_message",  data: { text: "帮我订机票" } },
  { type: "llm_call",      data: { model: "gpt-4o", prompt: "..." } },
  { type: "tool_call",     data: { tool: "search_flights", args: {...} } },
  { type: "tool_result",   data: { result: [...] } },
  { type: "llm_call",      data: { model: "gpt-4o", prompt: "..." } },
  { type: "tool_call",     data: { tool: "book_flight", args: {...} } },
  { type: "tool_result",   data: { result: { booking_id: "..." } } },
  { type: "agent_reply",   data: { text: "已为您预订..." } }
]

常见误区与权衡

🔗 延伸阅读