Metadata-Version: 2.1
Name: observation-agent
Version: 0.2.4
Summary: Agent 观测 SDK：模块四段（输入/结构/流程/输出）申报 + LangChain 自动采集 + HTTP 上报
Home-page: https://www.aiddit.com
Author: nieqi
Author-email: burningpush@gmail.com
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: langchain-core >=1.0
Provides-Extra: runtime
Requires-Dist: langchain >=1.0 ; extra == 'runtime'
Requires-Dist: langgraph >=1.0 ; extra == 'runtime'

# 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 **零第三方依赖**。

### 起观测台

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

### 接入方

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

```bash
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 档 · 申报（推荐）

```python
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 次执行）

```python
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()`。

### 确定性代码段

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

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

### 血缘

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

### 轮（循环型 Agent）

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

**一轮 = 锚点模块的第 K 次执行 → 第 (K+1) 次之前的全部关联执行。** 观测台多出轮选择器
`[全部][第1轮]…`，一轮一轮地看；跨轮容器每轮都在，作为结构框。锚点由业务声明、随每个 run
定死（业务最懂自己的轮语义），观测台只消费。不声明就没有轮。见 [参考 · round_anchor](obagent/docs/reference.md#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` 就是本次传进去的参数（唯一真值），两处不会漂移。

---

## 开发

```bash
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、自己开模块时，拓扑也别手抄：

```python
from obagent.integrations.langgraph import graph_spec

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

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