Agent 可观测性（四层观测模型）¶
本页用四层模型系统化 Eagle-RAG 的可观测性设计：如何在线上快速定位一次「agent 执行失败」—— 即一次走完 RAG 链路却结果异常的调用。四层均已实现；末尾给出 Langfuse 可选接入设计（未实现）。
边界声明
Eagle-RAG 是 RAG 数据层（ADR-008），不是 Agent 应用平台。本页的「agent 执行失败」指：(a) Eagle-RAG 自身链路失败（下游 agent 观测到错误/空结果/超时）；(b) Eagle-RAG 提供可被下游 agent 关联的 trace_id 上下文。实操端点与命令见 可观测性（运维）。
1. 四层模型总览¶
| 层 | 回答的问题 | 实现 | 定位失败的典型问题 |
|---|---|---|---|
| L1 Trace | 「失败发生在哪一步？」 | OpenTelemetry 追踪 | 某阶段 span ERROR / 耗时长 |
| L2 Metrics | 「是系统性还是单次？」 | Prometheus + metric_sample |
熔断、积压、延迟突增 |
| L3 Input/Output | 「这次具体输入输出是什么？」 | structlog ai_telemetry.jsonl |
prompt/completion/hits 异常 |
| L4 State | 「决策与状态怎么演变的？」 | PluginAudit + task_audit + documents.status |
路由错误、状态卡死、去重异常 |
核心关联键：trace_id 贯穿四层 —— L1 span、L3 JSONL 条目、L4 PluginAuditEvent 都带 trace_id/span_id（telemetry/context.py + logging_setup.add_open_telemetry_span 注入）。一次失败用一个 trace_id 即可串起四层证据。
2. Layer 1 — Trace 执行链路¶
代码： eagle_rag/telemetry/tracing.py
| 能力 | 函数 | 行为 |
|---|---|---|
| 追踪引导 | configure_tracing (L56) |
构建 TracerProvider（resource 含 service.name/version/environment）；按 tracing_enabled/otlp_endpoint 选 exporter：OTLP gRPC + BatchSpanProcessor（L92-97）/ ConsoleSpanExporter / NoOp |
| 开 span | trace_span (L109) |
双形态（上下文管理器 + 无参装饰器）；enter 时 bind_context(trace_id, span_id)；异常 record_exception + set_status(ERROR) |
| HTTP SERVER span | TelemetryMiddleware (L215) |
每请求开 SERVER span，提取 W3C traceparent 续接上游 trace，绑定 request_id；status>=500 置 ERROR |
| Celery CONSUMER span | register_celery_signals (L277) |
task_prerun 从 task.request.headers 提取父 span 开 CONSUMER span（名 {task.name}:{task_id}），绑定 job_id/document_id/kb_name；task_failure 记异常 |
| API→Celery 传播 | send_task_with_trace (L366) |
把当前 span 上下文 inject 到 Celery headers，跨进程续接 trace |
| GenAI 语义 | set_llm_span_attributes (L179) |
标注 gen_ai.system/gen_ai.request.model/gen_ai.prompt/gen_ai.completion/gen_ai.usage.*，prompt/completion 按 prompt_truncate(512)/completion_truncate(1024) 截断 |
/query 典型 trace 树¶
SERVER POST /query (TelemetryMiddleware)
├─ span route (QueryRouteClassifier)
├─ span retrieve (RetrieverOrchestrator)
│ ├─ span retrieve.text (per plan)
│ └─ span retrieve.visual (per plan)
├─ span rerank (rerank_merged)
└─ span generate (gen_ai.*) (EagleMultimodalQueryEngine)
异步入库则在另一条 trace：API SERVER → send_task_with_trace 投递 → Celery CONSUMER（ingest_router/knowhere_parse/pixelrag_build）。
默认 tracing 关闭
telemetry.tracing_enabled 默认 false（settings.yaml）。关闭时仍装 NoOp TracerProvider，trace_span 退化为 no-op 但仍生成有效 trace_id 供日志关联（logging_setup.add_open_telemetry_span 的 fallback）。开启需设 OTEL_TRACING_ENABLED=true + OTEL_EXPORTER_OTLP_ENDPOINT。
3. Layer 2 — Metrics 指标¶
代码： eagle_rag/metrics.py（MCP/Prometheus）+ eagle_rag/admin/metrics.py（metric_sample 时序表）
Prometheus 指标（/metrics 端点）¶
| 指标 | 类型 | 标签 | 含义 |
|---|---|---|---|
mcp_tool_calls_total |
Counter | tool, status |
MCP 工具调用计数；status = success/cache_hit/circuit_open/timeout/error |
mcp_tool_duration_seconds |
Histogram | tool |
工具调用耗时 |
mcp_active_requests |
Gauge | tool |
在途请求数 |
mcp_circuit_state |
Gauge | tool |
熔断状态 0=closed/1=half-open/2=open |
plugin_audit_decisions_total |
Counter | category, plugin_namespace, outcome |
插件决策计数；outcome = ok/error |
plugin_audit_rrf_dedupe_total |
Counter | plugin_namespace |
跨集合 RRF 去重事件（G32 双写监控） |
metric_sample 时序表（PG）¶
eagle_rag/db/models/metric_sample.py 定义通用时序采样表，由 admin/metrics.py 的 Celery beat 周期采样：
metric_name |
来源 | 用途 |
|---|---|---|
queue_size |
sample_queue_lengths（beat） |
各 Celery 队列深度 |
vlm_latency_ms |
生成路径采样 | VLM 延迟 |
vlm_tokens |
生成路径采样 | token 用量 |
vlm_error |
生成路径采样 | VLM 错误 |
GET /health 暴露队列时序（_read_queue_sizes，api/health.py L476）；get_metric_aggregate(name, agg, hours) 查聚合（如 vlm_latency_ms 的 24h avg，L1154）。
用指标定位：mcp_circuit_state=2 → 某工具熔断；queue_size 飙升 → worker 积压；mcp_tool_calls_total{status="error"} 速率上升 → 系统性故障而非单次。
4. Layer 3 — Input/Output¶
代码： eagle_rag/telemetry/logging_setup.py
| 能力 | 函数 | 行为 |
|---|---|---|
| AI 事件 logger | get_ai_logger(name) (L255) |
structlog BoundLogger，绑 component=name，写 logs/ai_telemetry.jsonl（telemetry.ai_log_file）；懒代理，telemetry 关闭时退回 stdlib |
| trace 注入 | add_open_telemetry_span (L196) |
structlog processor：从 OTel span 注入 trace_id/span_id/parent_span_id；无 span 时从 contextvars 取或生成随机 hex，保证每条日志可关联 |
| 截断 | truncate(text, limit) (L348) |
prompt/completion/query/hits 超 prompt_truncate(512)/completion_truncate(1024) 截断并标 ...<truncated>，防 JSONL 膨胀与敏感数据全量持久化 |
| 运维 logger | get_logger(name) (L266) |
loguru logger.bind，每次读 contextvars 保持 trace_id 最新；多 sink：stderr（pretty）+ 轮转文件（JSON）+ Redis pubsub |
热点埋点事件覆盖 route / retrieve / rerank / generate / ingest / mcp_call / query（详见 可观测性（运维） §AI 事件 logger）。每条 JSONL 带 trace_id/span_id/session_id/query_id/kb_name（经 bind_context 绑定）。
Redis pubsub sink（_make_redis_sink，L314）把 {level, message, timestamp, trace_id?} 发到 redis_log_channel（默认 logs），供 /admin/logs SSE 实时订阅。
5. Layer 4 — State 状态变化¶
代码： eagle_rag/plugins/audit.py（PluginAudit）+ eagle_rag/db/repositories/task_audit.py + documents.status 迁移
PluginAudit：多 sink 决策审计¶
PluginAudit.log_decision（audit.py L200）是稳定调用点，四路 fan-out（全部 best-effort，永不抛入调用方）：
| Sink | 实现 | 用途 |
|---|---|---|
| 内存 ring | deque(maxlen=ring_cap) (L145)，默认 1000 |
进程内最近窗口，Redis 不可用时 fallback |
| Redis LIST | LPUSH + LTRIM (L231)，key eagle:plugin_audit:{ns}:recent |
跨进程最近窗口 |
| AI JSONL | _emit_ai_log (L246) via get_ai_logger |
持久化、可聚合 |
| Prometheus | _emit_metrics (L275) |
聚合看板 |
PluginAuditEvent（L97）字段：category / target_collection / confidence / reason / plugin_namespace / kb_name / document_id / error / trace_id / span_id / extra。reason=="rrf_dedupe" 时额外增 plugin_audit_rrf_dedupe_total（与 证据聚合 去重审计衔接）。
其他状态源¶
task_audit仓库：Celery 任务审计持久化；documents.status迁移：入库状态机PENDING → RENDERING → EMBEDDING → INDEXING → SUCCESS（可靠性）；collections_usedcatalog：实际命中的集合记录。
暴露端点¶
GET /health/plugins 暴露 recent_decisions（PluginAudit.recent()，newest-last）与 audit_stats()（{buffer_size, source, enabled, redis_enabled}）。
6. 失败定位 Playbook¶
给定一个失败的 agent / RAG 调用（如下游 agent 报「查询返回空」或超时）：
1. 拿 trace_id
- 下游 agent 若传了 W3C traceparent，Eagle-RAG 会续接（TelemetryMiddleware L237 extract）
- 否则从 /admin/logs SSE 或响应头/日志拿 Eagle-RAG 生成的 trace_id
2. L1 Trace 找 ERROR span —— 定位失败阶段
- 在 OTel 后端按 trace_id 筛选；找 status=ERROR 或耗时异常的 span
- 无后端时：grep trace_id 走 L3 JSONL（见下），按 event 顺序还原阶段
3. L3 Input/Output 看具体 IO —— 判断是输入问题还是处理问题
grep <trace_id> logs/ai_telemetry.jsonl | jq .
- 看 retrieve 事件的 hits 是否为空（召回失败）
- 看 generate 事件的 prompt/completion（截断到 512/1024）
4. L2 Metrics 看是否系统性 —— 排除单次抖动
- /metrics 看 mcp_circuit_state 是否 = 2（熔断）
- /admin/celery 看 queue_size 是否积压
- GET /health 看 vlm_latency_ms / vlm_error 趋势
5. L4 State 看决策与状态 —— 找路由/分类错误
- GET /health/plugins 看 recent_decisions
- 看 category / target_collection / reason 是否误路由
- 看 documents.status 是否卡在某个中间态
实操命令（与现有端点一致）：
# 按 trace_id 还原一次调用的全部 AI 事件
grep <trace_id> logs/ai_telemetry.jsonl | jq .
# 队列与时序
curl /admin/celery # 队列深度（_read_queue_sizes）
curl /health/plugins # recent_decisions + audit_stats
curl /metrics # Prometheus 原始指标
7. Langfuse 可选接入设计（未实现）¶
本节为未实现的设计。Langfuse 当前未集成：grep -ri langfuse 仅命中 .trae/specs/integrate-ai-telemetry-tracing/spec.md 的「非目标」段。本设计给出基于现有 OTel 的零侵入接入点，落地需另起 ADR。
7.1 接入原理¶
Langfuse 原生支持 OTLP HTTP 摄取（https://cloud.langfuse.com/api/public/otel，Basic Auth = base64(public_key:secret_key)，header 另带 x-langfuse-ingestion-version=4）。关键映射：
- 带
gen_ai.*属性的 span → Langfuse generation； - 其余 span → Langfuse 普通 observation（trace/span 树）。
Eagle-RAG 已经通过 set_llm_span_attributes（tracing.py L179）在生成 span 上标注 gen_ai.*，且全链路有 OTel span。因此接入 Langfuse 无需改业务代码 —— 只需把 span 导出到 Langfuse 的 OTLP 端点。
7.2 接入点（两选一）¶
方案 A：加 OTLP HTTP exporter 分支（推荐）¶
configure_tracing（tracing.py L56）当前用 gRPC exporter：
# 现状（tracing.py L92-96）
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(endpoint=tel.otlp_endpoint, insecure=tel.otlp_insecure)
pyproject.toml 依赖 opentelemetry-exporter-otlp 包同时含 gRPC 与 HTTP exporter，故 HTTP 分支无需新依赖：
# 设计（未实现）：Langfuse 分支
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(
endpoint=tel.langfuse_endpoint, # https://cloud.langfuse.com/api/public/otel
headers={
"Authorization": f"Basic {base64(f'{pk}:{sk}')}",
"x-langfuse-ingestion-version": "4",
},
)
方案 B：OTel Collector sidecar¶
部署 OTel Collector，Eagle-RAG 仍发 gRPC 到 Collector，Collector 做 gRPC→HTTP 转发并注入 Langfuse Auth header。适合多服务统一出口。
7.3 Eagle-RAG span → Langfuse observation 映射¶
| Eagle-RAG span | Langfuse observation | 备注 |
|---|---|---|
POST /query（SERVER） |
trace 根 | |
route / retrieve / rerank |
span | |
generate（含 gen_ai.*） |
generation | 自动带 model/tokens/prompt/completion |
ingest_router / knowhere_parse（CONSUMER） |
span | 异步入库 trace |
7.4 Langfuse score 接入点（与 citationware 衔接）¶
Langfuse 的 score 机制可承载质量评测，与 Citationware RAG roadmap 衔接：
- faithfulness：每条 claim 是否被 cited source 支撑；
- citation 覆盖率：答案中无引文 claim 比例；
- abstain 率：证据不足时的显式拒答比例。
可在生成后用 Langfuse Python SDK 的 @observe / start_as_current_observation 显式增强 input/output 并 score() 上报（可选，非必需 —— 纯 OTLP 接入已能看到 trace 树）。
7.5 所需配置旋钮（设计层面，不落代码）¶
# settings.yaml telemetry 段（设计，未实现）
telemetry:
langfuse:
enabled: ${LANGFUSE_ENABLED:-false}
public_key: ${LANGFUSE_PUBLIC_KEY:-}
secret_key: ${LANGFUSE_SECRET_KEY:-}
endpoint: ${LANGFUSE_OTLP_ENDPOINT:-https://cloud.langfuse.com/api/public/otel}
落地时建议新增 ADR-009 记录「为何选 OTLP HTTP 直连 vs Collector sidecar」「score 评测范围」等决策。
8. 参考¶
- 可观测性（运维实操） —— 端点、命令、配置细节
- 可靠性（降级）
- 证据聚合（rrf_dedupe 审计）
- Citationware RAG（Langfuse score 衔接）
- ADR-008（RAG 数据层边界）
- 已有 telemetry spec（
.trae/specs/integrate-ai-telemetry-tracing/） - 术语表：Trace/Span / GenAI 语义约定 / PluginAudit / Langfuse / Observation