📡 可观测与追踪

class BaseCallbackHandler:
    """所有回调处理器的基类。定义全套生命周期钩子。"""

    # LLM 生命周期
    def on_llm_start(self, serialized, prompts, *, run_id, parent_run_id, **kwargs): ...
    def on_llm_new_token(self, token, *, run_id, **kwargs): ...   # 流式 token
    def on_llm_end(self, response, *, run_id, **kwargs): ...
    def on_llm_error(self, error, *, run_id, **kwargs): ...

    # Chain 生命周期
    def on_chain_start(self, serialized, inputs, *, run_id, ...): ...
    def on_chain_end(self, outputs, *, run_id, ...): ...

    # Tool 生命周期
    def on_tool_start(self, serialized, input_str, *, run_id, ...): ...
    def on_tool_end(self, output, *, run_id, ...): ...

    # Retriever 生命周期
    def on_retriever_start(self, ...): ...

    # 每个 run 都有 run_id 和 parent_run_id —— 构成父子 run 树！

# pip install langchain langchain-openai
from langchain_core.callbacks import BaseCallbackHandler
from langchain_openai import ChatOpenAI

class CostMonitorHandler(BaseCallbackHandler):
    """统计每次 LLM 调用的 token 成本。"""

    def __init__(self, pricing: dict):
        self.pricing = pricing          # {"gpt-4o-mini": {"input": 0.15, "output": 0.6}}（每百万 token $）
        self.total_input = 0
        self.total_output = 0
        self.total_cost = 0.0
        self.calls = 0

    def on_llm_end(self, response, *, run_id, parent_run_id, **kwargs):
        """每次 LLM 调用结束时，累加 token 和成本。"""
        self.calls += 1
        for generation in response.flatten():
            msg = generation.generations[0][0].message
            usage = getattr(msg, "usage_metadata", None)
            if usage:
                model = "gpt-4o-mini"  # 实际从 llm_output 取
                inp, out = usage.get("input_tokens", 0), usage.get("output_tokens", 0)
                self.total_input += inp
                self.total_output += out
                cost = (inp * self.pricing[model]["input"] +
                        out * self.pricing[model]["output"]) / 1_000_000
                self.total_cost += cost
                print(f"  [调用{self.calls}] in={inp} out={out} cost=${cost:.4f}")

    def report(self):
        print(f"\n===== 成本报告 =====")
        print(f"总调用: {self.calls}")
        print(f"总 token: in={self.total_input} out={self.total_output}")
        print(f"总成本: ${self.total_cost:.4f}")

# 使用：把 callback 通过 config 注入
handler = CostMonitorHandler(pricing={"gpt-4o-mini": {"input": 0.15, "output": 0.6}})
model = ChatOpenAI(model="gpt-4o-mini")

# 方式①：通过 config 的 callbacks 注入
result = model.invoke(
    "讲个笑话",
    config={"callbacks": [handler]},
)
handler.report()

# 方式②：构造时绑定（对该 model 的所有调用生效）
# model = ChatOpenAI(model="gpt-4o-mini", callbacks=[handler])

class LangChainTracer(BaseTracer):
    """上报 run 到 LangSmith 的 tracer。

    __init__ 接收 project_name（:142）；
    _persist_run（:281）把 Run 上报到 LangSmith。
    """
    def __init__(self, project_name=None, ...): ...

# 最简集成：设置环境变量，全自动追踪
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_..."
os.environ["LANGSMITH_PROJECT"] = "my-agent"

# 之后所有 invoke 自动追踪！无需改代码
agent.invoke({"messages": [...]})
# 去 LangSmith 网页能看到完整 run 树：
# 哪个 LLM 调用、调了什么 prompt、用了多少 token、工具执行多久

# types.py:133 注释：debug = checkpoints + tasks
StreamMode = Literal[..., "debug", ...]  # :120

# debug 模式：透出 Pregel 内部的 task 调度 + checkpoint 变化
# 由 pregel/debug.py 的 map_debug_tasks / map_debug_checkpoint 映射

for event in app.stream(inputs, stream_mode="debug"):
    if "tasks" in event:
        print(f"任务调度: {event['tasks']}")
    if "checkpoint" in event:
        print(f"状态快照: {event['checkpoint']['channel_values']}")
# 这是 LG5 Pregel 引擎最底层的可观测——看每个 super-step 内部

// wrap 装饰器（OC3 讲过）在工具执行外包 OTel span
function wrap(id, def, truncate, agents) {
  return async (args, ctx) => {
    // ...校验...
    return await def.execute(args, ctx).pipe(
      Effect.orDie,
      // ★ 每个工具调用一个 OTel span，带属性
      Effect.withSpan("Tool.execute", {
        attributes: {
          "tool.name": id,
          "session.id": ctx.sessionID,
          "message.id": ctx.messageID,
          "tool.call_id": ctx.callID,
        }
      })
    )
  }
}
// span 上报到 OTel collector → Jaeger/Grafana 可视化
// 比 LangSmith 更通用（不绑定特定平台）

// EventV2Bridge 服务：包装核心 EventV2，给事件附加 location（实例/工作区/项目）
// 订阅核心事件转发到 GlobalBus（UI 消费）
// durable 事件还会转成 sync 事件（跨设备同步）

// status.ts:39 会话状态变化时发布 SessionStatusEvent
// 这是面向 UI 的实时状态推送（idle/busy/retry）