AI Agent 从 demo 走向生产，最大的坑不是模型能力不够，而是你很快会发现：对话状态散落各处、决策路径无从回溯、用户断线重连后历史全丢。Apache Burr（目前处于 Apache 孵化阶段）就是来填这些坑的——它用一套零依赖的 Python 库，把 Agent 的状态机、执行追踪和持久化做成一件可以直接上手的事。
Agent 生产的三个硬伤
写一个能跑的 Agent 原型不难——调 API、拼 prompt、加几轮循环，半小时搞定。但一旦要上线，三件事立刻卡住你：
状态散乱。 Agent 每一步都可能修改上下文：用户偏好、工具返回值、中间推理结果。这些状态如果靠全局变量或 ad-hoc dict 管理，调试时你根本不知道"此刻的状态是怎么来的"。
决策不可追溯。 Agent 选了哪个工具、为什么跳过某一步、哪轮推理出了偏差——这些信息在日志里通常是零碎的字符串，事后拼不出来。
持久化断裂。 用户中途离开再回来，对话历史要能接上；服务重启后，正在执行的 Agent 不能丢状态。自己写存储逻辑容易，但和状态管理耦合在一起就越来越乱。
Burr 的思路是：把 Agent 当成显式状态机来写，每一步的输入输出和状态变更都自动记录，持久化只需换一个存储后端。
Burr 的核心模型：Action + State
Burr 的编程模型很简洁——你写的是一个个 Action，每个 Action 接收当前 State，返回新的 State 和可选的 Result。Application 对象负责串联这些 Action，驱动状态流转。
一个 Action 的最小实现：
from burr.core import Action, State, ApplicationBuilder, default
class AnalyzeIntent(Action):
def run(self, state: State) -> dict:
user_input = state["user_input"]
# 这里调用你的 LLM 或规则引擎
intent = "search" if "查" in user_input else "chat"
return {"intent": intent}
def update(self, state: State, result: dict) -> State:
return state.update(**result)
关键点：
run只做计算，不碰原状态——纯函数风格，方便测试。update把 run 的结果写入状态，变更显式可见。- State 本质是一个不可变 dict，每次
update生成新版本，不会悄悄篡改旧值。
串联成完整 Agent：
app = (
ApplicationBuilder()
.with_state(user_input="", intent="", search_result="", response="")
.with_actions(
analyze_intent=AnalyzeIntent(),
search_database=SearchDatabase(), # 你自己实现的 Action
generate_response=GenerateResponse(),
)
.with_transitions(
("analyze_intent", "search_database", default), # 默认走这条
("analyze_intent", "generate_response", lambda s: s["intent"] == "chat"),
("search_database", "generate_response", default),
)
.with_entrypoint("analyze_intent")
.build()
)
with_transitions 定义了 Action 之间的跳转规则——你可以用条件函数控制分支，而不是在 Action 内部硬编码 if-else。这让流程图从代码里浮出来，一眼能看清 Agent 的决策拓扑。
追踪：每一步自动记录
Burr 内置了追踪能力。你不需要自己打日志，只要在构建 Application 时挂上 tracker：
from burr.tracking import LocalTrackingClient
tracker = LocalTrackingClient("my_agent_runs")
app = (
ApplicationBuilder()
.with_state(**initial_state)
.with_actions(**actions)
.with_transitions(**transitions)
.with_entrypoint("analyze_intent")
.with_tracker(tracker)
.build()
)
运行后，Burr 会把每个 Action 的：
- 输入 State 快照
- 输出 Result
- 新 State 的 diff
- 执行耗时
全部写入本地文件（默认 JSON 格式）。Burr 还提供了一个 Web UI（burr serve），可以直接在浏览器里查看每次运行的完整时间线——哪个 Action 跑了、状态怎么变、哪条分支被选中，一目了然。
这对调试和生产监控都很有用：出问题时，你不需要翻日志猜，直接看追踪数据。
持久化：换一个后端就行
Burr 把状态存储抽象成 StatePersister。默认用内存，生产环境换成数据库只需一步：
from burr.persistence import SQLLitePersister
persister = SQLLitePersister(db_path="agent_state.db")
app = (
ApplicationBuilder()
.with_state(**initial_state)
.with_actions(**actions)
.with_transitions(**transitions)
.with_entrypoint("analyze_intent")
.with_persister(persister)
.build()
)
挂上 persister 后，每一步状态变更都会自动落盘。用户断线重连时，用同一个 app_id 重建 Application，状态直接恢复：
# 恢复上次对话
app = (
ApplicationBuilder()
.with_persister(persister)
.with_tracker(tracker)
.with_actions(**actions)
.with_transitions(**transitions)
.with_entrypoint("analyze_intent")
.with_app_id("user_123_session_456") # 同一个 app_id = 同一个对话
.initialize_from( # 从持久化存储加载状态
persister,
resume_from_last_step=True,
)
.build()
)
# 继续执行
result = app.run(halt_after=["generate_response"])
目前 Burr 提供了 SQLite 和 PostgreSQL 两种 persister。自己实现也不难——只要继承 StatePersister，写 load 和 save 两个方法，Redis、MongoDB 都能接。
一个完整的最小示例
下面是一个可以直接跑的完整 Agent——模拟一个"意图识别 → 数据查询 → 回复生成"的三步流程。先安装 Burr：
pip install burr[starters] # 包含 tracking UI 等起步依赖
完整代码 demo_agent.py：
from burr.core import Action, State, ApplicationBuilder, default, expr
from burr.tracking import LocalTrackingClient
# --- Action 定义 ---
class AnalyzeIntent(Action):
def run(self, state: State) -> dict:
text = state["user_input"]
intent = "search" if any(kw in text for kw in ["查", "搜索", "找"]) else "chat"
return {"intent": intent}
def update(self, state: State, result: dict) -> State:
return state.update(**result)
class MockSearch(Action):
def run(self, state: State) -> dict:
# 模拟数据库查询，实际项目替换为真实调用
query = state["user_input"]
return {"search_result": f"[模拟结果] 关于'{query}'的3条记录..."}
def update(self, state: State, result: dict) -> State:
return state.update(**result)
class GenerateResponse(Action):
def run(self, state: State) -> dict:
if state["intent"] == "search":
body = f"为您找到：{state['search_result']}"
else:
body = f"收到您的消息：{state['user_input']}"
return {"final_response": body}
def update(self, state: State, result: dict) -> State:
return state.update(**result)
# --- 构建 Application ---
tracker = LocalTrackingClient("demo_runs")
app = (
ApplicationBuilder()
.with_state(user_input="", intent="", search_result="", final_response="")
.with_actions(
analyze_intent=AnalyzeIntent(),
mock_search=MockSearch(),
generate_response=GenerateResponse(),
)
.with_transitions(
("analyze_intent", "mock_search", expr("intent == 'search'")),
("analyze_intent", "generate_response", expr("intent == 'chat'")),
("mock_search", "generate_response", default),
)
.with_entrypoint("analyze_intent")
.with_tracker(tracker)
.build()
)
# --- 执行 ---
inputs = {"user_input": "帮我查一下最近的订单"}
result = app.run(halt_after=["generate_response"], inputs=inputs)
print(result["final_response"])
# 输出: 为您找到：[模拟结果] 关于'帮我查一下最近的订单'的3条记录...
运行后查看追踪：
burr serve # 启动 Web UI，浏览器打开 http://localhost:7241
在 UI 里你会看到这次运行的完整步骤链：analyze_intent → mock_search → generate_response，每一步的状态变化都有记录。
上手建议与边界
适合的场景： Burr 最适合"步骤明确、状态流转可建模"的 Agent——比如客服机器人、数据处理 pipeline、多轮工具调用链。如果你的 Agent 是纯自由对话、没有固定流程，Burr 的状态机模型反而会显得约束过多。
当前限制： 项目还在孵化期，API 可能变动；persister 目前只有 SQLite 和 PostgreSQL；tracking 的存储格式也在演进中。建议先在非核心路径试用，等 API 稳定后再大规模接入。
和 LangGraph 的关系： 两者都做 Agent 流程编排，但定位不同。LangGraph 深度绑定 LangChain 生态，抽象层次更高；Burr 零依赖、模型更朴素，适合想自己控制每一步细节的团队。如果你已经在用 LangChain 全家桶，LangGraph 更顺滑；如果你想要一个轻量、可插拔的状态框架，Burr 更合适。
快速验证清单：
- ✅ 用
ApplicationBuilder搭一个 3-5 步的 Agent，跑通基本流程 - ✅ 挂上
LocalTrackingClient，在burr serve里看追踪数据 - ✅ 换
SQLLitePersister，模拟断线恢复：同app_id重建 Application，确认状态接上 - ✅ 把 mock Action 替换成真实 LLM 调用或工具函数，验证生产可行性
Burr 不试图解决所有 Agent 问题，但它把"状态、追踪、持久化"这三件容易做乱的事收拢到了一个干净的接口里。对于正在把 Agent 从原型推向生产的团队，这是一个值得花半天时间试一下的工具。