答案不对的 5 种可能原因：
1. 意图识别错了(本该 FAQ 直出，却走了文档 RAG)
2. source 推断错了(该搜 HR 文档却搜了 IT 文档)
3. 检索召回了无关内容(Embedding 或 BM25 失效)
4. 上下文构建截断了关键信息(max_context_chars 太小)
5. LLM 生成了幻觉(Prompt 约束不够)

RAG 运行时                     LangSmith 平台
┌──────────────────┐          ┌─────────────────────┐
│ record_query_trace│  ───→   │ Trace 存储 + 过滤    │
│ (147行 adapter)   │  metadata│ Dataset 管理         │
│                  │          │ Evaluation 自动评分   │
│ langsmith_status()│          │ Annotation 人工标注   │
└──────────────────┘          └─────────────────────┘

# qa_core/observability/langsmith_adapter.py
def configure_langsmith_environment() -> None:
    settings = get_settings()
    os.environ.setdefault("LANGSMITH_TRACING", "true" if settings.langsmith_tracing else "false")
    if settings.langsmith_api_key:
        os.environ.setdefault("LANGSMITH_API_KEY", settings.langsmith_api_key)
    if settings.langsmith_project:
        os.environ.setdefault("LANGSMITH_PROJECT", settings.langsmith_project)
    if settings.langsmith_endpoint:
        os.environ.setdefault("LANGSMITH_ENDPOINT", settings.langsmith_endpoint)

def langsmith_enabled() -> bool:
    settings = get_settings()
    return bool(settings.langsmith_tracing and settings.langsmith_api_key)

def record_query_trace(
    *,
    trace_id: str,
    session_id: str,
    question: str,
    answer: str,
    hit_type: str,
    scenario,
    data_scope: dict[str, Any],
    source_filter: str | None,
    kb_version: str,
    rewritten_query: str | None,
    intent: dict[str, Any] | None,
    retrieval: dict[str, Any] | None,
    sources: list[dict[str, Any]],
    elapsed_ms: float,
    error: str | None = None,
) -> None:

metadata = {
    "trace_id": trace_id,           # 与项目内部 trace_id 一致，可跨系统关联
    "session_id": session_id,
    "scenario_id": ...,             # 当前业务场景
    "scenario_name": ...,
    "source_filter": ...,           # 前端选择的业务分类
    "effective_source": ...,        # 最终生效的 source 过滤
    "kb_version": ...,              # 知识库版本号
    "tenant_id": ...,               # 数据隔离四维
    "dataset_id": ...,
    "visibility": ...,
    "user_role": ...,
    "intent": ...,                  # 意图分类结果
    "intent_reason": ...,           # 意图判断原因(rule / llm_structured)
    "hit_type": ...,                # faq_direct / rag / insufficient_context
    "prompt_profile": ...,          # 使用的 Prompt 档位
    "question_category": ...,       # 风险类别(pricing / compliance / ...)
    "sources_count": ...,           # 引用来源数量
    "top_source_score": ...,        # 最高召回分数
    "first_token_ms": ...,          # 首 token 延迟
    "stage_timings_ms": ...,        # 各阶段耗时
    "slowest_stage": ...,           # 最慢阶段
    "elapsed_ms": ...,              # 总耗时
    "error": ...,                   # 错误信息(如有)
}

全量 Trace ──→ 过滤(error / no_sources / low_score)──→ Annotation(人工标注)
                                                              │
                                                              ↓
                    Gate 验收 ←── Evaluation(自动评分)←── Dataset(评估集)

Trace 负责发现问题，Annotation 负责定义正确答案，Dataset 负责沉淀问题，Evaluation 负责批量评分，Experiment 负责对比版本。

用户问题：HS 编码归类存在争议时能先按客户说法申报吗？

Trace 中看到：
- scenario_id = cross_border_risk
- hit_type = rag
- sources_count = 1
- top_source_score = 0.42
- prompt_profile = compliance_guard
- answer_preview 中没有出现“归类依据”“申报风险”“补充资料”

{
  "is_correct": false,
  "expected_source": "customs",
  "expected_hit_type": "rag",
  "expected_keywords": ["HS 编码", "归类依据", "申报风险", "补充资料"],
  "grading_notes": "答案没有说明不能仅按客户说法申报，也缺少风险提示"
}

日常 bad case 沉淀：LangSmith
变更前强制验收：本地 Gate
长期趋势对比：LangSmith Experiments

app.py 启动时
  └── configure_langsmith_environment()   ← 注入环境变量

Pipeline 执行中(qa_core/pipeline/runtime.py)
  └── RAGQueryContext.run_stage()         ← 每个阶段自动计时
  └── RAGQueryContext.retrieval_info      ← 累积检索诊断数据
  └── RAGQueryContext.mark_first_token()  ← 记录首 token 时刻

Pipeline 结束时(qa_core/pipeline/rag.py)
  └── finish_success() / finish_error()
      └── RAGQueryContext.record_trace()
          └── record_query_trace()        ← 汇总所有数据写入 LangSmith

总耗时 = 意图识别 + Embedding + Milvus 检索 + Reranker + Prompt 构建 + LLM 首 token + 流式生成

QPS ≈ 并发请求数 / 平均请求耗时 = 30 / 6 = 5 QPS

hey -n 1000 -c 20 http://192.168.88.100:8001/health

hey -n 200 -c 10 -m POST \
  -H "Content-Type: application/json" \
  -d '{"query":"新人入职需要完成哪些流程？","scenario_id":"enterprise_knowledge"}' \
  http://192.168.88.100:8001/api/retrieval/debug

import asyncio
import json
import time
import websockets

URL = "ws://192.168.88.100:8001/api/stream"
PAYLOAD = {
    "question": "新人入职需要完成哪些流程？",
    "scenario_id": "enterprise_knowledge",
}

async def one_user(i: int):
    started = time.perf_counter()
    first_token = None
    async with websockets.connect(URL, ping_interval=20) as ws:
        await ws.send(json.dumps(PAYLOAD, ensure_ascii=False))
        async for msg in ws:
            event = json.loads(msg)
            if event.get("type") == "token" and first_token is None:
                first_token = time.perf_counter() - started
            if event.get("type") in {"end", "error"}:
                total = time.perf_counter() - started
                return {"user": i, "first_token": first_token, "total": total, "type": event.get("type")}

async def main():
    results = await asyncio.gather(*(one_user(i) for i in range(30)))
    print(results)

asyncio.run(main())

代码变更
  → pytest 单元测试
  → 文档/配置一致性检查
  → 构建 API 镜像
  → 预发环境 rebuild_kb_version.py --quality-gate
  → 本地 Evaluation / LangSmith Evaluation
  → 小流量压测
  → 生产 docker compose --env-file .env.compose pull + up -d
  → 观察 LangSmith Trace、错误率、首 token、Milvus P95

docker compose --env-file .env.compose up -d --force-recreate api
docker logs -f knowforge-api

docker compose --env-file .env.compose run --rm api python scripts/rebuild_kb_version.py --scenario enterprise_knowledge --new-version --force --reset-collections --quality-gate --activate

1. 看服务是否活着：docker compose --env-file .env.compose ps / healthcheck
2. 看启动校验：preflight 是否全部通过
3. 看业务状态：scenario、active kb_version、collection 名称
4. 看 trace：命中路径、阶段耗时、top score、sources_count
5. 看依赖：Milvus/MySQL/LLM/Embedding/Reranker
6. 看数据：资料是否入库、source 是否正确、质量门禁是否通过

def configure_langsmith_environment() -> None:
    settings = get_settings()
    os.environ.setdefault("LANGSMITH_TRACING", "true" if settings.langsmith_tracing else "false")
    if settings.langsmith_api_key:
        os.environ.setdefault("LANGSMITH_API_KEY", settings.langsmith_api_key)
    if settings.langsmith_project:
        os.environ.setdefault("LANGSMITH_PROJECT", settings.langsmith_project)
    if settings.langsmith_endpoint:
        os.environ.setdefault("LANGSMITH_ENDPOINT", settings.langsmith_endpoint)

def record_query_trace(*, trace_id, session_id, question, answer, hit_type,
                       scenario, data_scope, source_filter, kb_version,
                       rewritten_query, intent, retrieval, sources,
                       elapsed_ms, error=None) -> None:
    configure_langsmith_environment()
    if not langsmith_enabled():
        return

    retrieval_payload = retrieval or {}
    intent_payload = intent or {}
    metadata = {
        "trace_id": trace_id,
        "session_id": session_id,
        "scenario_id": getattr(scenario, "scenario_id", ""),
        "scenario_name": getattr(scenario, "display_name", ""),
        "source_filter": source_filter,
        "effective_source": retrieval_payload.get("source_filter") or source_filter,
        "kb_version": kb_version,
        "tenant_id": data_scope.get("tenant_id"),
        "dataset_id": data_scope.get("dataset_id"),
        "visibility": data_scope.get("visibility"),
        "allowed_roles": data_scope.get("allowed_roles"),
        "intent": intent_payload.get("intent"),
        "intent_reason": intent_payload.get("reason"),
        "hit_type": hit_type,
        "prompt_profile": retrieval_payload.get("prompt_profile_name"),
        "question_category": retrieval_payload.get("question_category"),
        "rewritten_query": rewritten_query,
        "sources_count": len(sources),
        "top_source_score": sources[0].get("score") if sources else None,
        "first_token_ms": retrieval_payload.get("first_token_ms"),
        "stage_timings_ms": retrieval_payload.get("stage_timings_ms"),
        "slowest_stage": retrieval_payload.get("slowest_stage"),
        "elapsed_ms": round(elapsed_ms, 2),
        "error": error,
    }

    with trace("qa_stream_query", run_type="chain", metadata=metadata, run_id=trace_id) as run_tree:
        outputs = {"answer_preview": answer[:800], "hit_type": hit_type, "sources": sources[:8], "error": error}
        run_tree.end(outputs=outputs, error=error) if error else run_tree.end(outputs=outputs)

with context.run_stage("doc_retrieval"):
    doc_hits = retriever.search_many(queries)

with context.run_stage("llm_generation"):
    yield from llm.stream(prompt)

docker compose --env-file .env.compose up -d --build api
curl http://192.168.88.100:8001/health
python scripts/acceptance_smoke.py --base-url http://192.168.88.100:8001