agentocc

AgentOCC

LangGraph 多智能体共享状态的乐观一致性中间件 (OCC middleware)

⚠️ Scope Lock — 本项目只做一件事:Stale-Generation 异常的 commit-time 校验。 不做 MVCC / 悲观锁 / Serializable Snapshot Isolation / Tool-Effect Reordering 检测 / Write Skew 跨 key 校验 / 细粒度 replan。 详见 设计文档 §3。

---

这是什么

多智能体系统(LangGraph 式编排)中,多个 Agent 共享同一份全局状态。LLM 的生成阶段长达数秒到数分钟,把传统数据库里"读-提交"的毫秒级窗口拉长了几个数量级——于是 Agent A 读到的状态,在它生成期间可能已经被 Agent B 改掉。A 提交时仍基于旧值,产生 Stale-Generation:既不报错,也不被感知,只静默传播。

AgentOCC 在 LangGraph 现有 checkpoint 机制之上包一层拦截(不改内核),用经典乐观并发控制(OCC)的 Backward Validation 在提交时刻检测这类异常,冲突时整体重试。

本项目的核心论点(§4.3):不是照搬 OCC,而是实测 OCC 的"冲突很少发生"这一经典假设,在 LLM 长生成窗口下是否依然成立。

核心机制

经典 OCC 三阶段(Kung & Robinson, 1981)落到 Agent 执行:

四个核心组件

快速上手

from agentocc import AgentOCC

guard = AgentOCC()
guard.seed({"shared_value": 0})

@guard.agent(max_retries=3)
def my_agent(ctx):
    # Read phase — records version-at-read
    current = ctx.read("shared_value")
    # Generate phase — long LLM generation would happen here
    # Write phase — stages output, does not touch global state directly
    ctx.write("shared_value", current + 1)

tx = my_agent()
print(guard.read("shared_value"))
print(guard.metrics())

LangGraph 风格节点可以用 guard_node 包一层:

from agentocc import AgentOCC, guard_node

guard = AgentOCC()

def planner(ctx):
    task = ctx.read("task")
    return {"plan": f"Plan for {task}"}

node = guard_node(planner, guard=guard, reads=["task"], writes=["plan"])
updates = node({"task": "release the SDK"})

项目结构

agentocc/
├── agentocc/                  # 核心包(纯 stdlib,无强依赖)
│   ├── store.py                 # VersionedStateStore  组件 1
│   ├── transaction.py           # AgentTransaction     组件 2
│   ├── core.py                  # AgentOCCCore       组件 3
│   ├── runner.py                # run_agent_with_guard 组件 4
│   └── langgraph_integration.py # LangGraph checkpoint 适配(可选)
├── tests/                       # 单元测试
├── experiments/                 # 故障注入 + 量化指标(§6)
│   ├── scenarios/               #   2-3 个 Stale-Generation 触发场景
│   └── metrics.py               #   拦截率 / 延迟 / token 成本
├── examples/                    # 用法示例
└── docs/StateGuard_设计文档.md

实验指标(§6.2)

• 异常拦截率:构造的冲突场景中,被 AgentOCC 成功拦截的比例
• 延迟开销:加入校验层后相对无防护版本的平均延迟增量 (ms)
• Token 成本对比:重试导致的重新生成带来的 token 成本增量
• Baseline 对比:同样 query 集,无校验层直接执行的静默错误传播率

开发

pip install -e ".[dev]"
pytest

许可证

MIT