内容
# 状态波多智能体共享上下文
**多个智能体，一个真实来源。**
[](#需求)
[](#许可证)
> **状态波生态系统的一部分：** [服务器](https://github.com/smaramwbc/statewave) · [Python SDK](https://github.com/smaramwbc/statewave-py) · [TypeScript SDK](https://github.com/smaramwbc/statewave-ts) · [连接器](https://github.com/smaramwbc/statewave-connectors) · [文档](https://github.com/smaramwbc/statewave-docs) · [示例](https://github.com/smaramwbc/statewave-examples) · [网站 + 演示](https://statewave.ai) · [管理员](https://github.com/smaramwbc/statewave-admin)
>
> 📋 **问题和功能请求：** 在 [`smaramwbc/statewave`](https://github.com/smaramwbc/statewave/issues) 中集中跟踪 — 问题在此仓库中禁用，因此所有报告都集中在一个地方。
并行智能体读取和写入相同的权威上下文层可以防止彼此之间的冲突在发生之前。
## 目录
- [状态波多智能体共享上下文](#状态波多智能体共享上下文)
- [目录](#目录)
- [问题](#问题)
- [解决方案](#解决方案)
- [工作原理](#工作原理)
- [本仓库构建的内容](#本仓库构建的内容)
- [三个演示时刻](#三个演示时刻)
- [1. 碰撞（之前）](#1-碰撞之前)
- [2. 没有碰撞（之后）](#2-没有碰撞之后)
- [3. 共享时间线](#3-共享时间线)
- [截图](#截图)
- [需求](#需求)
- [技术栈](#技术栈)
- [快速入门](#快速入门)
- [1. 在本地启动状态波](#1-在本地启动状态波)
- [2. 克隆和安装此演示](#2-克隆和安装此演示)
- [3. 配置凭证](#3-配置凭证)
- [4. 运行演示](#4-运行演示)
- [项目结构](#项目结构)
- [状态波集成](#状态波集成)
- [故障排除](#故障排除)
- [为什么不使用 AutoGen / LangGraph？](#为什么不使用autogen-langgraph)
- [许可证](#许可证)
---
## 问题
在朴素的并行智能体系统中，规划者做出架构决策（例如 *弃用旧 session-token 模块*），并将其写入自己的上下文中。编码器同时开始工作，但从未看到该决策。编码器从头开始重建已弃用的模块。审查者在两个智能体都已完成工作后才捕获冲突 - 为时已晚，无法防止计算资源的浪费。
这是并行智能体系统中记录的真实故障模式。由于智能体通过消息传递上下文，而不是通过共享的持久层，因此会发生这种情况。
## 解决方案
[状态波](https://statewave.ai) 为每个智能体提供一个共享的**主题**（运行 ID）。决策作为**事件**写入，并编译为**类型化记忆**，每个智能体在行动之前都会读取这些记忆。
- 编码器读取规划者的弃用决策，然后再开始工作。
- 冲突永远不会发生。
- 可以从 `GET /v1/timeline` 重建完整的决策链。
## 工作原理
状态波是一个共享内存存储，位于智能体之间，而不是智能体之间直接传递消息。每个运行中的智能体都从同一个**主题**（由运行 ID标识）读取和写入，因此“是否有人已经决定了这一点？”是一个查找，而不是猜测。
数据模型分为三个层次：
1. **事件** 智能体记录的原始事实，例如 *"规划者决定弃用旧 session-token 模块。"* 使用 `sw_client.write_episode(...)` → SDK `create_episode()`写入。廉价、仅追加、无解释。
2. **编译** 将新写入的事件转换为可检索的记忆。使用 `sw_client.compile(...)` → SDK `compile_memories_wait()`调用。此步骤存在是因为原始事件在编译之前不可搜索/排名 - 跳过它会导致陈旧读取。
3. **记忆** 从一个或多个事件（例如 `architectural_decision`，置信度 `0.92`）派生的类型化、置信度评分、可查询的事实。其他智能体使用 `sw_client.get_context(...)` → SDK `get_context()`拉取相关记忆，这些记忆针对特定任务进行排名。
具体来说，在此仓库的 `--mode statewave` 管道中 ([pipeline_statewave.py](pipeline_statewave.py))：
```bash
规划者
→ 决定弃用 legacy-session-token 模块
→ write_episode(type="architectural_decision", content="Deprecation: legacy-session-token")
→ compile(subject_id) # 决策现在对任何人可见
编码器
→ get_context(subject_id, task="implement feature: ...") # 在编写任何代码之前
→ 看到规划者的弃用记忆在返回的上下文中
→ 跳过 → legacy-session-token 模块（由规划者弃用：使用 JWT）
→ write_episode(type="implementation_note", ...)
→ compile(subject_id)
审查者
→ get_context(subject_id, task="review implementation: ...")
→ 将规划者的决策与编码器的输出进行交叉检查
→ 发现没有冲突，因为编码器已经有了决策
```
这就是为什么冲突被**防止**而不是**检测**：编码器的上下文读取发生在它决定构建什么之前，而不是之后。在朴素的管道中 ([pipeline_naive.py](pipeline_naive.py))，没有等效的读取 - 编码器只会看到原始任务字符串，因此它无法知道规划者已经做出了使部分计划无效的决策。
运行的完整事件 → 编译 → 记忆链可以通过 `python timeline_inspector.py --run <id>` 进行端到端审计 - 参见下面的 [共享时间线](#3-共享时间线)。
## 本仓库构建的内容
一个由三个智能体组成的 Python CLI，运行模拟的软件开发任务：
| 智能体 | 角色 |
|----------|-------------------------------------------|
| 规划者 | 做出架构决策，包括弃用 |
| 编码器 | 实现功能 |
| 审查者 | 验证输出是否与真实来源一致 |
演示运行 **两个管道**，针对同一任务：
1. **之前** (`--mode naive`) 智能体仅使用消息传递。发生冲突。
2. **之后** (`--mode statewave`) 智能体共享一个状态波主题。防止冲突。
第三个命令 `python timeline_inspector.py --run <id>`，打印已完成运行的完整时间线审计 - 每个智能体在行动时知道什么，它写了什么，以及最终的记忆状态。
## 三个演示时刻
### 1. 碰撞（之前）
```
[规划者] 弃用 → legacy-session-token 模块 替换为 JWT
[编码器] 无共享上下文可用 仅限于任务描述。
[编码器] → legacy-session-token 模块
[编码器] → jwt-auth 模块
[审查者] ╔══ 碰撞检测 ═══════════════════════════════════════╗
║ 检测到冲突 但两个智能体都已完成 ║
║ 他们的作品。浪费的计算无法恢复。 ║
╚════════════════════════════════════════════════════════════╝
```
### 2. 没有碰撞（之后）
```
[规划者] 弃用 → legacy-session-token 模块 替换为 JWT
[规划者] 编译事件 → 使决策对所有智能体可用...
[编码器] 读取共享上下文，然后再开始实施...
[编码器] [architectural_decision]（置信度 0.92） 弃用：legacy-session-token
[编码器] 跳过 → legacy-session-token 模块（由规划者弃用：使用 JWT）
[审查者] ╔══ 冲突避免 ═════════════════════════════════════════╗
║ 审查完成 未发现冲突。 ║
║ 编码器在行动前读取了规划者的弃用决定。 ║
╚════════════════════════════════════════════════════════════╝
```
### 3. 共享时间线
```
$ python timeline_inspector.py --run sw-abc12345
时间（UTC） 相对 智能体 事件类型 内容
──────────────────────────────────────────────────────────────────────
14:01:00Z T+0.0s 规划者 architectural_decision 弃用：legacy-session-token...
14:01:12Z T+12.1s 规划者 compile 内存编译触发
14:01:15Z T+15.3s 编码器 context_retrieval 检索 2 条记忆（包括弃用）
14:05:00Z T+4m0s 编码器 implementation_note 已构建：jwt-auth、rbac、login、logout
14:05:30Z T+4m30s 审查者 context_retrieval 检索 4 条记忆
14:05:45Z T+4m45s 审查者 review_finding 状态：CLEAN
```
---
## 截图
[工作原理](#工作原理) 中的架构图 ([docs/images/architecture.svg](docs/images/architecture.svg)) 是此仓库的规范视觉表示 - 它是一个 SVG，以状态波自己的深靛蓝/紫色主题构建（与上面的横幅匹配），因此它在任何尺寸下都保持清晰，并且可以作为文本文件轻松编辑，如果管道发生变化。
其他演示截图和录制内容位于 [docs/images/](docs/images/)。添加一个：
1. 将图像文件放入 `docs/images/`（例如 `docs/images/collision-detected.png`）。
2. 使用 `` 从此 README 中引用它 - 参见文件顶部横幅图像的精确语法（文件名中的空格必须编码为 `%20`）。
3. 特别是对于终端输出，优先粘贴原始文本在围栏代码块中（在 [三个演示时刻](#三个演示时刻）中）而不是截图 - 它保持可复制和可区分，并且如果 `core/display.py` 样式更改，则不会过时。
---
## 需求
- **Python 3.11+**
- **Node.js** （用于 `npx @statewavedev/statewave`）或 Docker，如果自行托管状态波
- **LLM 提供商 API 密钥** 演示通过 [LiteLLM](https://docs.litellm.ai) 调用模型，因此任何支持的提供商都有效。默认提供商是 **Groq**（慷慨的免费层） 在 [console.groq.com/keys](https://console.groq.com/keys) 获取密钥
## 技术栈
- **[LiteLLM](https://docs.litellm.ai)** 提供商无关的 LLM 调用。默认模型：`groq/llama-3.3-70b-versatile`。指向任何提供商，设置 `LLM_MODEL` 为 `<provider>/<model>`（例如 `openai/gpt-4o-mini`，`anthropic/claude-3-5-haiku-20241022`，`ollama/llama3`）。包装在 [core/llm.py](core/llm.py) 中。
- **状态波** （官方 [`statewave`](https://pypi.org/project/statewave/) Python SDK） 共享上下文层（事件、编译的记忆、时间线）。参见 [core/statewave_client.py](core/statewave_client.py)。
- **Rich** 终端输出
---
## 快速入门
### 1. 在本地启动状态波
状态波是自行托管的。使用一个命令启动它，它启动 API + Postgres via Docker：
```bash
# macOS / Linux
npx @statewavedev/statewave
# Windows（PowerShell）
irm https://www.statewave.ai/install.ps1 | iex
```
API 将在 `http://localhost:8100` 可用。无需帐户或 API 密钥。
### 2. 克隆和安装此演示
```bash
git clone https://github.com/smaramwbc/statewave-multi-agent-shared-context
cd statewave-multi-agent-shared-context
pip install -r requirements.txt
```
### 3. 配置凭证
```bash
cp .env.example .env
# 编辑 .env 添加您的 LLM_API_KEY（免费 Groq 密钥：https://console.groq.com/keys）
# STATEWAVE_BASE_URL 默认设置为 http://localhost:8100 无需更改
```
### 4. 运行演示
```bash
# 完整的 before/after 比较（默认）
python main.py
# 仅限 naive 管道
python main.py --mode naive
# 仅限 Statewave 管道
python main.py --mode statewave
# 自定义任务
python main.py --task "构建支付服务。弃用旧 Stripe v1 模块。"
# 检查已完成运行的共享时间线
python timeline_inspector.py --run sw-abc12345
```
---
## 项目结构
```
statewave-multi-agent-shared-context/
├── main.py # CLI 入口点 协调两个管道
├── timeline_inspector.py # 时间线检查命令
├── pipeline_naive.py # BEFORE：智能体仅使用消息传递
├── pipeline_statewave.py # AFTER：智能体共享一个 Statewave 主题
├── agents/
│ ├── planner.py # 规划者智能体（naive + Statewave 模式）
│ ├── coder.py # 编码器智能体（naive + Statewave 模式）
│ └── reviewer.py # 审查者智能体（naive + Statewave 模式）
├── core/
│ ├── statewave_client.py # 适配器官方 statewave SDK
│ ├── llm.py # LiteLLM 支持的客户端（OpenAI 样式调用表面）
│ └── display.py # 基于 Rich 的终端帮助程序
├── statewave_agents/ # 可重用的 SharedContext 原语（框架无关）
│ ├── context.py # SharedContext：before_acting() / decide()
│ └── agent.py # 可复用 SharedContext 原语（框架无关）
│ ├── agent.py # 智能体基类
├── requirements.txt
└── .env.example
```
---
## 状态波集成
所有状态波调用都通过官方 [`statewave`](https://pypi.org/project/statewave/) Python SDK 进行 没有手写的 HTTP 请求。
| 操作 | SDK 方法 | 何时 |
|-----------|------------|------|
| 写决策 | `create_episode()` | 每次规划者决策后 |
| 编译 | `compile_memories_wait()` | 立即写入后 使内存可用（处理多批次排水） |
| 读取上下文 | `get_context()` | 任何智能体行动前 |
| 检查时间线 | `get_timeline()` | 审计 / `timeline_inspector.py` |
每个调用都包含一个 `caller_id`，因此整个决策链可归因于每个智能体。
---
## 故障排除
**`litellm.AuthenticationError`**
您的 `LLM_API_KEY` 缺失或与 `LLM_MODEL` 中的提供商不匹配。密钥必须属于提供商前缀 Groq 密钥用于 `groq/...`，OpenAI 密钥用于 `openai/...` 等。
**`litellm.BadRequestError` / "LLM 提供商未提供" / 模型未找到**
`LLM_MODEL` 必须是完全限定的 `<provider>/<model>` 字符串（例如 `groq/llama-3.3-70b-versatile`），而不是裸模型名称。参见 [LiteLLM 的提供商列表](https://docs.litellm.ai/docs/providers).
**Windows 上的 `UnicodeEncodeError`（'charmap' 编码无法编码字符 '→'）**
Windows 控制台编码（cp1252）无法呈现用于 `core/display.py` 中的 Unicode 箭头。强制使用 UTF-8：
```powershell
$env:PYTHONIOENCODING = "utf-8"; python main.py
```
**`localhost:8100` 连接被拒绝**
状态波服务器未运行。参见 [快速入门步骤 1](#1-在本地启动状态波)。
---
## 为什么不使用 AutoGen / LangGraph？
AutoGen 群聊和 LangGraph 分层智能体教程都通过 **消息** 路由上下文。消息按顺序传递状态。它们不为所有智能体提供在同一时刻访问相同的权威来源。
- 在顺序管道中，消息传递有效，因为智能体轮流行动。
- 在并行管道中，没有顺序。消息发送时，接收智能体已经开始或已经完成。
状态波用共享的读写表面替换协调层。无需布线。无需二次消息开销。每个智能体读取相同的事实真相。
状态波是自行托管的 没有托管云 API。参见 [快速入门步骤 1](#1-在本地启动状态波) 以在本地运行。运行后，文档位于 `http://localhost:8100/docs`。源代码：[github.com/smaramwbc/statewave](https://github.com/smaramwbc/statewave)。
---
## 与其他智能体框架集成
此仓库中的协调原语与 `agents/*.py` 无关 它是 [`SharedContext`](statewave_agents/context.py)，它只需要围绕智能体已经执行的操作调用两个方法：
- `context.before_acting(caller_id, task)` 调用前调用，查看其他智能体已经做出了什么决定。
- `context.decide(caller_id, content, kind=...)` 调用后调用，发布决策供其他人查看。
任何框架的智能体循环都可以围绕这两个调用。`caller_id` 和 `subject_id`（运行/会话 ID）是您需要携带的仅有的两个概念。
### CrewAI
将相同的原语公开为 CrewAI `Tool`，因此任何 Crew 智能体都可以作为其正常工具调用循环的一部分读写共享主题：
```python
from crewai import Agent, Crew, Task
from crewai.tools import tool
from statewave_agents.context import SharedContext
shared = SharedContext(subject_id="crew-run-001")
@tool("check_shared_decisions")
def check_shared_decisions(task_description: str) -> str:
"""在行动前，查看其他智能体已经做出了什么决定。"""
bundle = shared.before_acting(caller_id="crew-agent", task=task_description)
return bundle.assembled_context or "No prior decisions yet."
@tool("publish_decision")
def publish_decision(decision: str, kind: str = "decision") -> str:
"""发布决策，以便其他智能体可以看到。"""
shared.decide(caller_id="crew-agent", content=decision, kind=kind)
return "Published."
planner = Agent(role="Planner", goal="...", tools=[check_shared_decisions, publish_decision])
coder = Agent(role="Coder", goal="...", tools=[check_shared_decisions, publish_decision])
Crew(agents=[planner, coder], tasks=[...]).kickoff()
```
给 Crew 中的每个智能体赋予两个工具和相同的 `subject_id` CrewAI 仍然负责任务排序，而 Statewave 负责“是否有人已经知道这一点？”
### Claude 智能体 SDK
将相同的两个操作注册为 SDK 工具，以便任何 Claude 智能体都可以在其调用过程中调用它们：
```python
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient
from statewave_agents.context import SharedContext
shared = SharedContext(subject_id="claude-run-001")
@tool("check_shared_decisions", "查看其他智能体已经做出的决定", {"task": str})
async def check_shared_decisions(args):
bundle = shared.before_acting(caller_id="claude-agent", task=args["task"])
return {"content": [{"type": "text", "text": bundle.assembled_context or "No prior decisions yet."}]}
@tool("publish_decision", "发布决策供其他智能体查看", {"decision": str, "kind": str})
async def publish_decision(args):
shared.decide(caller_id="claude-agent", content=args["decision"], kind=args.get("kind", "decision"))
return {"content": [{"type": "text", "text": "Published."}]}
server = create_sdk_mcp_server(name="shared-context", tools=[check_shared_decisions, publish_decision])
options = ClaudeAgentOptions(mcp_servers={"shared-context": server})
async with ClaudeSDKClient(options=options) as client:
await client.query("Implement the login endpoint.")
```
运行每个角色的 Claude 智能体，都指向同一个 MCP 服务器实例（相同的 `subject_id`），以便它们读写同一个主题。
### 任何其他框架或自定义框架（LangGraph、AutoGen、手工制作的循环等）
没有特定于 SDK 的步骤 每个框架的智能体都有 *某个* 时候要采取行动和 *某个* 时候产生输出的地方。在这两个地方准确地挂钩 `SharedContext`:
```python
from statewave_agents.context import SharedContext
shared = SharedContext(subject_id=run_id)
def run_my_agent(caller_id: str, task: str):
bundle = shared.before_acting(caller_id, task) # 1. 行动前读取
result = my_framework_agent.run(task, extra_context=bundle.assembled_context)
shared.decide(caller_id, result.decision, kind=result.kind) # 2. 行动后写入
return result
```
唯一的要求：运行中的每个智能体都必须共享相同的 `subject_id`，并且每个写入操作都必须发生在下一个智能体的读取操作之前，以便读取操作有所作用 排序保证，而不是框架，保证防止冲突。
---
## 许可证
Apache 2.0 参见 [LICENSE](LICENSE)。
MCP 配置
以下是此 MCP Server 的配置信息，可直接复制到 Cursor 或其他 MCP 客户端中使用。
mcp.json
连接信息
猜你喜欢
暂无推荐服务器