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 上,随时可查、可回放。

基本信息

核心 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 实现对比

特性Republiclossless-clawTape (设计)
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 调用的审计和上下文管理:

参考资料