author: Bub Build contributors (PsiACE, Frost Ming, Yihong)
source: https://getrepublic.org/
tags: [agent, ai, python, tape, llm, audit, openclaw]
rating: ⭐⭐⭐⭐⭐ (5/5)
Republic — Tape 优先的 LLM 客户端库
一句话版本
Republic 是一个 Python 库,让你像写普通 Python 代码一样调用 LLM(聊天、工具、流式),同时自动记录完整的审计轨迹——每次对话、每次工具调用、每个错误都作为不可变的结构化数据存到 Tape 上,随时可查、可回放。
基本信息
- 官网: getrepublic.org
- 仓库: bubbuild/republic
- 协议: Apache-2.0
- 语言: Python 3.12+
- 安装:
pip install republic - 源起: 衍生自 lightning-ai/litai,受 pydantic/pydantic-ai 启发
- 定位: Bub 的底层 Tape 存储和 LLM 调用层
核心 Design
"Tape-First" 哲学
Republic 的核心设计理念是 tape-first:每条消息、工具调用、工具结果、错误、用法统计都以结构化数据记录在 Tape 上。开发者先让工作流可审计,再决定哪里需要添加智能。
from republic import LLM
llm = LLM(model="openrouter:openrouter/free", api_key="<API_KEY>")
tape = llm.tape("ops")
# 写入 anchor + 聊天,全部自动记录
tape.handoff("incident_42", state={"owner": "tier1"})
out = tape.chat("Connection pool is exhausted. Give triage steps.", max_tokens=96)
# 全量审计
print([entry.kind for entry in tape.query.all()])
# → ['anchor', 'message', 'tool_call', 'tool_result', 'message']
四种核心原语
| 原语 | 描述 | 代码 |
|---|---|---|
| **LLM** | 统一入口(chat/tool/stream/embedding) | `LLM(model=..., api_key=...)` |
| **Tape** | Append-only 执行日志 | `llm.tape("name")` |
| **StructuredOutput** | 返回值 + 错误(永远不会抛异常) | `out.value` / `out.error` |
| **ToolExecutor** | 自动或手动的工具执行 | `executor.run(...)` |
TapeEntry 类型
Republic 定义了 7 种 immutable Entry:
| Kind | 用途 | 字段 |
|---|---|---|
| `message` | 一条 LLM 消息 | content, role |
| `system` | 系统提示 | content |
| `anchor` | 阶段标记 | name, state |
| `tool_call` | 工具调用 | calls |
| `tool_result` | 工具执行结果 | results |
| `error` | 错误记录 | kind, message, trace |
| `event` | 自定义事件 | name, data |
TapeQuery 链式查询
# 查询所有条目
tape.query.all()
# 按关键词 + 类型 + 时间范围组合过滤
tape.query.query("timeout").kinds("message").between_dates("2026-03-01", "2026-03-07").limit(10).all()
# 基于 Anchor 的上下文切片
tape.query.after_anchor("incident_42").all()
# 最近一次 Anchor 之后
tape.query.last_anchor().all()
# 两个 Anchor 之间
tape.query.between_anchors("start_phase", "end_phase").all()
技术架构
源码结构
src/republic/
llm.py — LLM 统一入口
tape/ — Tape 核心
__init__.py — 公开 API
entries.py — TapeEntry(7 种 kind,immutable dataclass)
query.py — TapeQuery(fluent builder,支持链式 + overload)
session.py — Tape(llm.tape(name) 返回的 session 对象)
manager.py — AsyncTapeManager / TapeManager
store.py — 存储引擎(InMemoryTapeStore, AsyncTapeStoreAdapter)
context.py — TapeContext(指定 anchor 位置构建上下文)
clients/ — LLM 客户端
chat.py — 聊天客户端
embedding.py — 嵌入客户端
text.py — 文本补全
github_copilot.py — Copilot 客户端
openai_codex.py — Codex 客户端
parsing/ — 响应解析
tools/ — 工具执行
executor.py — ToolExecutor(自动/手动模式)
schema.py — 工具 Schema 定义
context.py — 工具上下文
auth/ — 认证
openai_codex.py — OpenAI Codex OAuth
github_copilot.py — GitHub Copilot OAuth
core/ — 核心类型
errors.py — RepublicError(结构化错误)
results.py — StructuredOutput(value + error 模式)
client_registry.py — 客户端注册
execution.py — 执行引擎
关键设计决策
1. StructuredOutput 模式 — 返回值封装为 value + error,永远不抛异常,retry/fallback 逻辑可确定性地写
2. Sync / Async 双模式 — 同步 API 用 chat(),异步用 chat_async(),底层存储自动适配
3. TapeStore 抽象 — InMemoryTapeStore 用于测试,AsyncTapeStoreAdapter + 自定义存储(如 SQLite 或云存储)用于生产
4. 两种 Tape 查询路径 — tape.query.*() 走 store 原生查询,TapeContext 用于构建 LLM 上下文消息
5. Anchor-based 上下文切片 — 通过 after_anchor("name") 或 between_anchors("a", "b") 精确控制上下文窗口
与 tape.systems 的对应
| tape.systems 概念 | Republic 实现 |
|---|---|
| Entry | `TapeEntry(kind, payload, meta, date)` |
| Anchor | `TapeEntry.anchor(name, state)` 通过 `tape.handoff()` |
| Append | `tape.append_async(TapeEntry.message(...))` |
| View | `TapeQuery` 链式过滤,配合 `TapeContext` |
| Handoff | `tape.handoff(name, state=...)` |
| Corrections | 追加新 entry,不覆写历史 |
| Fork/Merge | `InMemoryTapeStore` 与 `TapeManager` 的 fork 支持 |
在 Bub 中的使用
Bub 的 builtin/tape.py 直接使用 Republic 的 Tape API:
from republic import LLM, AsyncTapeStore, Tape, TapeEntry, TapeQuery
# Bub 的 TapeService
class TapeService:
def __init__(self, llm, archive_path, store):
self._llm = llm
self._store = store
async def info(self, tape_name):
tape = self._llm.tape(tape_name)
entries = list(await tape.query_async.all())
# 统计 anchors、token 用量等
async def handoff(self, tape_name, *, name, state=None):
tape = self._llm.tape(tape_name)
return await tape.handoff_async(name, state=state)
async def search(self, query):
return list(await self._store.fetch_all(query))
评分对比
| 维度 | 评分 | 说明 |
|---|---|---|
| **Tape 实现完整度** | ⭐⭐⭐⭐⭐ | 完整实现了 tape.systems 的全部核心概念 |
| **接口简洁度** | ⭐⭐⭐⭐⭐ | `llm.chat()` 一行调用,自动审计 |
| **链式查询** | ⭐⭐⭐⭐⭐ | TapeQuery 的 fluent API 设计优雅 |
| **双模式支持** | ⭐⭐⭐⭐ | sync + async 都有,但 async 模式限制较多 |
| **存储引擎** | ⭐⭐⭐ | 只有 InMemory 内置,自定义存储需自己实现 |
| **文档质量** | ⭐⭐⭐⭐⭐ | getrepublic.org 文档完整,有示例代码 |
Tape 实现对比
| 特性 | Republic | lossless-claw | Tape (设计) |
|---|---|---|---|
| Append-only | ✅ | ✅ | ✅ |
| Anchor | ✅ handoff() | ✅ (sum_xxx) | ✅ |
| Handoff | ✅ | ❌ 仅 compact | ✅ |
| Fork/Merge | ✅ InMemoryTapeStore | ✅ DAG | ✅ |
| Query | ✅ 链式 fluent | 🔍 grep | 概念 |
| Context 构建 | ✅ TapeContext | 内置 | View |
| 事件/错误 | ✅ 一等公民 | ❌ | ✅ |
深度分析
惊艳点
1. Tape-first 做得彻底 — 不是"加个审计功能",而是所有 LLM 交互天生记录在 Tape 上
2. StructuredOutput 模式 — 返回值永远不抛异常,out.value / out.error 让 retry 逻辑像普通 if 语句
3. Fluent 查询设计 — tape.query.query("x").kinds("y").between_dates("a","b").all() 读起来像自然语言
4. 同组织复用 — 被 Bub 直接依赖为 Tape 引擎,且其他项目也可以独立使用
5. 错误作为一等公民 — 错误是 TapeEntry,不是异常,可以被查询、回放和分析
6. 衍生自 litai — 站在巨人肩膀上,结构干净
不足/风险
1. 只有 InMemory 存储 — 内置存储引擎只有内存版,生产级需要自己实现 TapeStore
2. 异步模式限制 — sync tape API 在 async store 下不可用(会 raise),限制较多
3. 小众生态 — 与 Bub 深度绑定,独立使用场景有限
4. 文档较新 — 仍在完善中,部分 API 有 deprecation 标记
5. Agent 能力缺失 — 专注 LLM 调用和审计,不提供 Agent 框架级别的能力(技能、子 Agent 等)
对我们项目的意义
Republic 是最纯粹的 Tape 设计语言实现——不包含任何 Agent 框架的额外复杂度,只关注 LLM 调用的审计和上下文管理:
- 与 lossless-claw 互补思路:lossless-claw 的 DAG 摘要压缩先前的上下文,Republic 的 Anchor-based 切片选择上下文。两者可以结合使用
- StructuredOutput 模式值得借鉴到 OpenClaw 的工具调用中
- 链式查询 API可以作为 lossless-claw grep 工具的用户体验参考
- 如果开发 Python 版本的 OpenClaw Agent,Republic 是理想的 LLM 层和 Tape 层