observation-agent 0.2.4

Agent 观测 SDK：模块四段（输入/结构/流程/输出）申报 + LangChain 自动采集 + HTTP 上报

obagent · Agent 运行时与观测台

给 LangChain / LangGraph Agent 用的观测 SDK + 观测台。和 Langfuse / LangSmith 那类 「上报 + 可视化」的区别只有一条，但决定了一切：

它们看到的是 span 树；这里看到的是「模块四段」。

span 树能从 callback 里自动扒出来 —— 所以谁都能做，也就不构成差异化。 四段（① 输入 · ② 结构 · ③ 运行流程 · ④ 输出）扒不出来，必须由作者申报。 于是整个工程目标只有一个：把申报的成本压到接近零。

---

核心分工

	谁产生	内容
自动	LangChain callback	模型推理正文 / 思考 / tool_calls / token、工具入参出参、时序与并发批次、消息原件、system prompt 文本、模型参数
申报	只有作者知道	模块身份与边界、输入由哪几个槽构成、输出是什么与成败、确定性代码段、扇出的哪一路是哪一路

一句话记法：作者知道而机器猜不到的才申报；机器能看见的一律自动。

为什么身份最要紧：callback 给的是每次调用一个随机 UUID。没有稳定的 module_key， 定义去重 → 注册表 → 版本对比（「改了这句 prompt 到底变好没有」）→ 标注按模块聚合，全塌。

---

两个东西，分得很清

	谁装	依赖	职责
obagent（SDK）	每个接入方	只有 langchain-core	申报 + 自动采集 + HTTP 上报
server/（观测台）	只有观测台这台机器	+ SQLAlchemy / PyMySQL / FastAPI	ingest 接口 + 查询接口 + 前端 + 数据库

SDK 不碰数据库。 它会被 pip 装进任意业务进程，持有生产库口令等于把口令散布出去。 观测数据一律经 HTTP 交给服务端。上报走 stdlib urllib，SDK 零第三方依赖。

起观测台

pip install -r requirements-server.txt
# ~/.obagent/config.json 的 db 段填连接串（见 config.example.json）
python -m server.app --init --port 8931      # --init 幂等建库建表

接入方

完整接入文档：obagent/docs/ —— 按主题分页（快速开始 / 申报 / LangGraph / 按轮看 / API 参考…），例子都自包含可直接跑。 文档随 SDK 一起装，装完直接 python -m obagent docs 离线看（docs <篇名> / --grep 关键词 / --export），不用手工拷贝。

pip install observation-agent
export OBAGENT_ENDPOINT=http://观测台地址:8931
export OBAGENT_PROJECT=你的项目名          # 可选，module_key 的第一段

python -m obagent doctor      # 地址配了吗？观测台连得上吗？有没有积压的 WAL？
python -m obagent replay      # 上报失败时暂存的数据，补投

没配 OBAGENT_ENDPOINT → 不落任何数据并告警一次，业务代码照常跑。 这是刻意的：观测挂了不该把业务带崩，但绝不能静默 ——「以为记下来了其实没有」比没记更糟。

上报为什么不阻塞业务

Store 协议是「插入后返回 id」的（call 边要指向被调实例）。走 HTTP 就意味着每建一行一个往返。 所以改成客户端分配 uid（和 OpenTelemetry 的 span id 一个路子）：SDK 立刻返回自己造的 i7 / n23，把「uid → 真实主键」的解析留给服务端。于是所有写入都只进队列、不等回包。

三条硬要求：不阻塞主流程（后台单线程批量发）· 不丢数据（发不出去落本地 WAL 并告警）· 不改时序语义（单队列严格保序，批量窗口 ≤0.5s，「边跑边落」的实时性保住）。

用法

1 档 · 申报（推荐）

from obagent import observe
from obagent.observe import InputBlock

with observe.run(agent="my_agent", project="acme", objective="回答用户问题"):

    with observe.module("router", kind="agent", title="意图路由") as ctx:
        user = ctx.declare(                       # ← 返回拼好的 user_content
            system_prompt=SYS,
            blocks=[InputBlock("用户问题", q, key="question", optional=False),
                    InputBlock("历史", history, key="history")],
            tools=TOOLS)

        out = my_own_graph.invoke({"messages": [("user", user)]})   # 任何 LangChain 代码
        ctx.set_output(out, ok=True)

declare() 返回真正喂给模型的那段文本 —— 申报不是额外负担，它就是「拼 prompt」那一步。 同一份 blocks 既拼 prompt 又落视图，展示与真实调用永不漂移。

0 档 · 零申报

只包一层 observe.run(...)，模型/工具调用照样被采集，模块在画面上标 「未申报结构」。 用来先看见，再逐步往 1 档走。

扇出（一份定义 × N 次执行）

def check(h, dim):
    h.declare(system_prompt=EVAL, blocks=[InputBlock("本维度", dim, key="dim")])
    ...
    h.set_output(verdict, ok=True)
    return verdict

verdicts = ctx.map("item", check, dims, keys=dims, titles=dims)

N 路共享一个 fanout_group、各自一个 branch_key、同一个 seq（= 同时发起）， 且开线程之前就建好 N 行 —— 执行期间前端能看到 N 个盒子同时亮起、各自填充。

⚠️ 自己开线程时别用裸 executor.submit —— contextvars 不会自动跟过去，父帧会丢。 用 observe.run_parallel(...) / observe.spawn(executor, fn)，它们内部 copy_context()。

确定性代码段

ctx.record_stage("split", fn=split_dimensions, output={"dimensions": dims}, ok=True)

fn 给了就把源码本身收进结构 —— 确定性代码没有 system prompt 可看，能回答 「它按什么规则工作」的只有代码；而且源码进指纹 ⇒ 改代码即新版本。

血缘

product = gen.out("url")                                  # 带来源引用的值
InputBlock("被评产物", key="cand", value=product)          # ← origin 自动带上

轮（循环型 Agent）

with observe.run(agent="planner", objective="…", round_anchor="plan"):
    app.invoke(state)                    # ← 只多这一个参数

一轮 = 锚点模块的第 K 次执行 → 第 (K+1) 次之前的全部关联执行。 观测台多出轮选择器 [全部][第1轮]…，一轮一轮地看；跨轮容器每轮都在，作为结构框。锚点由业务声明、随每个 run 定死（业务最懂自己的轮语义），观测台只消费。不声明就没有轮。见 参考 · round_anchor。

---

可视化原则

查看器面向通用设计：它只还原两样东西 —— 模块的结构与本次的运行流程。 它不认识任何业务概念（没有 Task 卡、没有成品图廊、没有硬编码的业务轮次面板）。

一切显示都来自产生侧写进字段的事实：卡片标题、成败、并发列、被调实例、结构化产物。 这一层不做任何语义推断 —— 一旦允许它猜（按名字猜这是个 Task、按 URL 猜这是成品图）， 查看器就绑死在某一个业务上了，而这套东西的立身之本恰恰是通用。

轮选择器是这条原则的一个正例：它确实存在，但不猜任何业务 —— 一轮的边界由业务自己在 observe.run(round_anchor=…) 里声明（某个模块的每次执行为一轮），查看器只按声明切分、不硬编码 「plan 就是一轮」这种业务知识。不声明就没有轮，整个 run 摊在一张画板。

业务定制的可视化是另一层的事（渲染器注册表），后续再谈。

数据模型

一切皆模块。模块 = 四段。模块之间只有引用，没有嵌套。

oa_module_def   模块定义（全局，跨 run 复用）  身份 = module_key + fingerprint(版本)
oa_module_ref   定义 → 定义（结构里引用了谁）
oa_module_inst  运行实例（一次执行）           无 parent、无 path
oa_flow_node    运行节点 llm/tool/note/call    call 即「实例 → 实例」的边
oa_message      消息原件（dumpd，可逐字还原）
oa_annotation   旁路标注（重放 context 后模型自述依据）

三条不变量：

1. 四种节点共用同一个 seq 序号空间 ⇒「想了想 → 调了个子模块 → 再想了想」排得出顺序；
2. 同 seq + 同 fanout_group = 同时发起 ⇒ 并发不是新层级，是同层兄弟关系；
3. 没有链可爬 ⇒ task_id 这类字段每行显式落。

调用链就靠一列：oa_flow_node.callee_inst_id。call 节点不存入参副本 —— 被调实例的 input 就是本次传进去的参数（唯一真值），两处不会漂移。

---

开发

pip install -e .                         # SDK
pip install -r requirements-server.txt   # 观测台
python -m server.app --init &            # 起服务（含建表）

python test/run_all.py                   # 全部用例（末尾一张总表）
python test/run_all.py graph_agent       # 只跑某一类
python -m server.projection.flatcheck <run_id>   # 一致性检查 + 可读投影

打开 http://127.0.0.1:8931 看画板。

用例 = 接入说明书

每个用例都是能真跑的完整程序（一律假模型：不花钱、不要 key、结果可复现），跑完自己 校验一致性并打印「去哪看 / 看什么」。挑一个长得像你的代码的照抄，比读文档快：

用例	这种代码形态怎么接
test/code/01_basic.py	单个确定性函数：源码即结构、签名即输入槽
test/code/02_loop.py	循环与递归：一份定义 N 个实例、递归是嵌套不是并发
test/code/03_pipeline.py	多段管道 + 并发 + 血缘
test/agent/01_decorator_vs_with.py	注解写法 vs with 写法（产出等价，按代码形态选）
test/agent/02_runtime_call.py	函数体只有 observe.call()（真调模型，--paid）
test/graph_agent/01_zero_declare.py	LangGraph 0 档：一行申报都不写能看到什么
test/graph_agent/02_declared.py	同一张图申报之后，多出来的是什么
test/graph_agent/03_plan_execute.py	plan-execute 循环图：回边 · 一份定义 N 次执行 · 并发验收
test/unit/test_core.py	16 条不变量回归（不连库）

只要那张图

不用 wrapper、自己开模块时，拓扑也别手抄：

from obagent.integrations.langgraph import graph_spec

with observe.module("loop", kind=observe.KIND_WORKFLOW, spec=graph_spec(app)) as ctx: ...

文档：接入文档（obagent/docs/）（给使用者，也可 python -m obagent docs） · 系统实现（给读懂/改动 obagent 本身的人）