Observability 可观测与评估

pip install langfuse

# .env
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_BASE_URL=https://cloud.langfuse.com   # 自托管换成你的域名

from langfuse.langchain import CallbackHandler

handler = CallbackHandler()

result = graph.invoke(
    {"messages": [("user", "什么是 RRF 融合？")]},
    config={"callbacks": [handler], "thread_id": "chat-9001"},
)

print(handler.last_trace_id)   # 拿到 trace id，用于关联工单与用户反馈

from langfuse import observe

@observe()                                  # 顶层调用成为 trace
def answer(query: str) -> str:
    docs = retrieve(query)                  # 子 observation 自动嵌套
    return compose(query, docs)

@observe()
def retrieve(query: str) -> list[str]:
    return vector_db.similarity_search(query, k=5)

@observe(name="compose", as_type="generation")   # 标记为 LLM 调用节点
def compose(query: str, docs: list[str]) -> str:
    return llm.invoke(build_prompt(query, docs)).content

from langfuse import get_client

langfuse = get_client()

with langfuse.start_as_current_observation(name="answer") as trace:
    docs = do_retrieve(query)          # 内部再开的 observation 自动成为子节点
    answer_text = do_generate(query, docs)
    trace.update(output=answer_text)   # 手动补输出；异常也会被记录

from langfuse import get_client, propagate_attributes

langfuse = get_client()

def handle_request(user_id: str, session_id: str, query: str) -> str:
    with langfuse.start_as_current_observation(name="answer") as trace:
        with propagate_attributes(
            user_id=user_id,
            session_id=session_id,        # 建议直接用 LangGraph 的 thread_id
            tags=["production", "rag"],
        ):
            answer_text = answer(query)   # 内部所有埋点自动继承上下文
            trace.update(output=answer_text)
            return answer_text

from langfuse import get_client

langfuse = get_client()

langfuse.create_prompt(
    name="support-answer",
    type="chat",
    prompt=[
        {"role": "system", "content": "你是 {{product}} 的客服，用{{tone}}语气回答，答不了就明说。"},
        {"role": "user", "content": "{{question}}"},
    ],
    labels=["production"],   # 创建即可上线；也可以先不挂 label
)

prompt = langfuse.get_prompt("support-answer", type="chat")
messages = prompt.compile(product="云数据库", tone="专业简洁", question=query)

resp = llm.invoke(messages)

{
  "input": {
    "question": "退款多久到账？",
    "contexts": ["...检索到的片段，可选..."]
  },
  "expected_output": "原路退回，1-3 个工作日到账",
  "metadata": {
    "source": "trace-a1b2c3",
    "failure_type": "格式漂移"
  }
}

from langfuse import get_client, Evaluation

langfuse = get_client()
dataset = langfuse.get_dataset("support-golden")

def answer_task(*, item, **kwargs):
    # item.input / item.expected_output / item.metadata 全部可用
    return my_app(item.input["question"])

def correctness(*, input, output, expected_output, **kwargs):
    ok = expected_output.strip() in output
    return Evaluation(
        name="correctness",
        value=1.0 if ok else 0.0,
        comment="命中黄金要点" if ok else "未命中",
    )

result = dataset.run_experiment(
    name="hybrid-rerank-v2",          # run 名：分支名或 PR 号
    description="混合检索 + 重排，对比基线",
    task=answer_task,
    evaluators=[correctness],
)
print(result.format())               # 汇总：每个评估器的均值与失败清单

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

class Faithfulness(BaseModel):
    """判断回答是否完全由给定上下文支撑"""
    supported: bool
    reason: str = Field(description="指出第一个越界论断；没有则写'无'")

judge_llm = init_chat_model("openai:gpt-4o")
faithful = judge_llm.with_structured_output(Faithfulness)

def faithfulness(*, input, output, **kwargs):
    contexts = "\n".join(input.get("contexts", []))
    verdict = faithful.invoke(
        f"上下文：\n{contexts}\n\n回答：\n{output}\n\n"
        "回答中的每个论断都能由上下文支撑吗？"
    )
    return Evaluation(
        name="faithfulness",
        value=1.0 if verdict.supported else 0.0,
        comment=verdict.reason,
    )

from langfuse import get_client
from langfuse.langchain import CallbackHandler

# 路由层：执行时取到 trace_id，随响应返回给前端
handler = CallbackHandler()
result = graph.invoke(
    {"messages": [("user", query)]},
    config={"callbacks": [handler], "thread_id": session_id},
)
response = {"answer": result["messages"][-1].content,
            "trace_id": handler.last_trace_id}

# API 层：用户点击赞/踩后，把评价补写到同一条 trace
@app.post("/feedback")
def feedback(trace_id: str, rating: int, comment: str | None = None):
    get_client().score(
        trace_id=trace_id,
        name="user-feedback",
        value=rating,          # 1 = 踩, 2 = 一般, 3 = 赞
        comment=comment,
    )