用 Langfuse 给 Agent 加可观测与评估：Trace、Prompt 版本与数据集回归

# 方式 A：Langfuse Cloud（零运维，有免费额度）
#   注册后建项目，直接拿 key，跳到 Step 2

# 方式 B：Docker 自托管（数据留在自己机器上）
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d
# 默认 Web 控制台在 http://localhost:3000
# 底层组件：PostgreSQL（事件）、ClickHouse（分析）、Redis（队列）、MinIO（对象存储）

pip install langfuse

# .env（不要提交到 git）
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_BASE_URL=https://cloud.langfuse.com
# 其它区域端点（按你的账号区域选择）：
#   美国: https://us.cloud.langfuse.com
#   日本: https://jp.cloud.langfuse.com

如果你用的是 OpenAI SDK，Langfuse 提供了一个「即插即用」的封装：把 import 换掉，其余写法不变，调用就会被自动记录。

# 原来是：from openai import OpenAI
# 现在换成 Langfuse 的封装（行为一致，额外记录每次调用）
from langfuse.openai import openai

completion = openai.chat.completions.create(
    name="first-trace",                     # 给这次调用起个名字，便于检索
    model="gpt-4o",                         # 模型名以官方文档为准
    messages=[
        {"role": "system", "content": "你是一个只输出计算结果的计算器。"},
        {"role": "user", "content": "1 + 1 = ?"},
    ],
    metadata={"env": "local"},              # 自定义元数据，方便筛选
)
print(completion.choices[0].message.content)

跑完去控制台看 Trace 列表，应该能看到这一条，点进去能看到输入、输出、模型、token 与耗时。

真实 Agent 里除了模型调用，还有检索、工具、后处理。用 @observe() 装饰器可以把它们都挂进同一条链路，形成父子 Span：

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse()  # 读取环境变量里的 key 与端点

@observe(as_type="generation")
def call_llm(messages, model="gpt-4o"):
    """被装饰后，输入/输出/耗时/token 会被自动捕获。"""
    resp = openai.chat.completions.create(model=model, messages=messages)
    langfuse_context.update_current_observation(
        model=model,
        usage={
            "input": resp.usage.prompt_tokens,
            "output": resp.usage.completion_tokens,
        },
    )
    return resp.choices[0].message.content

@observe()
def run_agent(user_input: str):
    """顶层函数：它下面的所有 @observe 调用都会成为子 Span。"""
    langfuse_context.update_current_trace(
        user_id="user-123",
        session_id="session-abc",     # 同一会话的多轮会归到一起
        tags=["production"],
    )
    answer = call_llm([{"role": "user", "content": user_input}])
    return answer

if __name__ == "__main__":
    print(run_agent("用一句话解释什么是 RAG"))

调试时踩过的坑，应该变成下次的测试用例。Langfuse 的路径是：

1. 把提示词放到控制台管理：改一次生成一个版本，线上引用具体版本；出问题时能精确回滚到「上一版」。避免「改了 prompt 但没人记得改了啥」。
2. 建数据集：把「信息缺失」「含无关材料」「来源冲突」这几类难例各存几条，作为回归集。
3. 批量跑评估：改动前后各跑一遍同一数据集，对比答案质量、token、延迟，用数据决定是否上线。