
CubePi 是一个 Pythonic 原生异步 Agent 框架,专为高性能、高可读性和生产级持久化而设计。 它以线性 while 循环建模 agent 逻辑,提供比图结构 agent 运行时更轻量的替代方案, 开发者可以轻松追踪和调试。
| langgraph | CubePi | |
|---|---|---|
| 抽象模型 | 图节点 + 边 + 通道 | 普通 async 函数 — run_agent_loop 就是一个 while 循环 |
| 流式输出 | 基于回调,多种 handler 类型 | async for event in stream — 统一模式 |
| 检查点 | 每步全量快照,序列化整个消息列表 | 追加式 — 无论对话多长,每轮 O(1) DB I/O |
| 依赖项 | langchain-core、langgraph-sdk 及传递依赖 | 3 个核心依赖:pydantic、anthropic、openai |
| 工具执行 | 工具是需要手动连线的图节点 | 声明为函数;框架自动路由并并行执行 |
| 多 Provider | 通过 langchain chat model 适配器 | 原生 Provider 协议 — 内置 Anthropic、OpenAI |
| Middleware | 图级中间件,在节点进出时触发 | 8 种类型化 hook,声明式组合规则 |
| 可观测性 | LangSmith / Langfuse 集成 | 原生 OpenTelemetry — Tracer、Meter、GenAI semconv,开箱即用 OTLP / JSONL |
import asyncio
from cubepi import Agent, tool
from cubepi.providers.anthropic import AnthropicProvider
provider = AnthropicProvider(api_key="sk-...")
@tool
async def get_weather(city: str) -> str:
"Get current weather for a city."
return f"72°F and sunny in {city}"
agent = Agent(
model=provider.model("claude-sonnet-4-6"),
tools=[get_weather],
system_prompt="You are a helpful weather assistant.",
)
def on_event(event, signal=None):
if event.type == "text_delta":
print(event.delta, end="", flush=True)
agent.subscribe(on_event)
asyncio.run(agent.prompt("What's the weather in Tokyo?"))
pip install cubepiuv add cubepipoetry add cubepipip install cubepi[sqlite,postgres,mcp,tracing,tracing-otlp]