feat: 接入langfuse

.env.example

@@ -15,3 +15,12 @@ INTENT_MIN_CONFIDENCE=0.7
 
 # 动作自动修正次数，不含首次生成，不限制用户澄清轮数
 MAX_ACTION_RETRIES=2
+
+# Langfuse 追踪 不影响流程
+# 未设置 LANGFUSE_TRACING_ENVIRONMENT 时代码默认 development
+LANGFUSE_SECRET_KEY="sk-xxx"
+LANGFUSE_PUBLIC_KEY="pk-xxx"
+LANGFUSE_BASE_URL="http://127.0.0.1:3000"
+LANGFUSE_TRACING_ENVIRONMENT=development
+# 设为 false，关闭 langfuse 追踪
+LANGFUSE_TRACING_ENABLED=true

app/clients/decision.py

@@ -77,6 +77,11 @@ class DecisionEngine:
             | {"type": question.type}
             for name, question in request.decisions.items()
         }
+        from app.observability.tracing import observe_decision, record_decision
+
+        # 观测包在请求外面：成功才记结果，失败仍抛原来的 DecisionError
+        # generation 为 None 时 record_decision 什么都不做
+        with observe_decision(request) as generation:
             try:
                 response = await client.system_one(
                     state=request.state,
@@ -104,7 +109,9 @@ class DecisionEngine:
                 raise DecisionError("无法连接 Jev 服务", kind="connection") from error
             except TypeSafeError as error:
                 raise DecisionError("Jev SDK 调用失败", kind="sdk") from error
-        return self._parse_response(request, response)
+            parsed = self._parse_response(request, response)
+            record_decision(generation, parsed)
+            return parsed
 
     @staticmethod
     def _parse_response(

app/observability/tracing.py

@@ -0,0 +1,284 @@
+"""一次大屏图调用的 Langfuse 追踪
+
+不属于处理管道的任一步。把图调用收成可过滤的 trace，并把 Jev 记成当前节点下的 generation
+每轮调用单独记录，首次请求与澄清恢复通过 session 关联，不修改业务状态
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from asyncio import CancelledError
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+from app.config import settings
+
+logger = logging.getLogger("app.observability")
+
+# 面板 Trace Name 列用的稳定名字，不要拼 thread_id 或用户原文
+TRACE_NAME = "screen-instruct"
+# 按功能过滤用的 tag，不是每一次请求的动态标签
+FEATURE_TAG = "screen-agent"
+# 客户端复用连接；callback 按调用创建，不保留跨请求的恢复关联
+_client: Any = None
+_unavailable = False
+
+
+def tracing_enabled() -> bool:
+    """三项连接配置都有，且没有显式关掉，才发送 trace
+
+    LANGFUSE_TRACING_ENABLED=false 用来本地临时关闭，缺任何一项连接配置也视为关闭
+    """
+    flag = os.getenv("LANGFUSE_TRACING_ENABLED", "").strip().lower()
+    if flag in {"0", "false", "no"}:
+        return False
+    return bool(
+        os.getenv("LANGFUSE_PUBLIC_KEY", "").strip()
+        and os.getenv("LANGFUSE_SECRET_KEY", "").strip()
+        and os.getenv("LANGFUSE_BASE_URL", "").strip()
+    )
+
+
+def flush_tracing() -> None:
+    """把还在内存里的 span 立刻发出去
+
+    Langfuse 默认攒一批再发。演示脚本马上退出时不调用，这条 trace 会丢
+    追踪没启用或客户端没建成时直接返回，不报错
+    """
+    client = _client
+    if client is None or _unavailable:
+        return
+    try:
+        client.flush()
+    except Exception:
+        logger.warning("Langfuse flush 失败", exc_info=True)
+
+
+async def ainvoke_screen(
+    graph: Any,
+    payload: Any,
+    *,
+    thread_id: str,
+    session_id: str | None = None,
+) -> Any:
+    """每轮调用记录一个根 span，返回、等待用户或异常时结束。
+
+    session_id 由调用方在首次与恢复时一致传入（通常为 conversation_id）。
+    未传时始终使用 thread_id，不从仅首次请求才有的 payload 推断。
+    """
+    config = {"configurable": {"thread_id": thread_id}}
+    client = _ensure_client() if tracing_enabled() else None
+    if client is None:
+        return await graph.ainvoke(payload, config=config)
+
+    from langfuse import propagate_attributes
+    from langfuse.langchain import CallbackHandler
+    from langgraph.types import Command
+
+    handler = CallbackHandler()
+    # 在当前异步上下文执行回调，手动记录的 Jev generation 才会挂到所属节点。
+    handler.run_inline = True
+    resume = isinstance(payload, Command) and payload.resume is not None
+    trace_input = (
+        {"reply": payload.resume}
+        if resume
+        else {"prompt": payload.get("request", {}).get("prompt")}
+    )
+    with (
+        client.start_as_current_observation(
+            name=TRACE_NAME,
+            as_type="span",
+            input=trace_input,
+        ) as root,
+        propagate_attributes(
+            trace_name=TRACE_NAME,
+            session_id=session_id or thread_id,
+            tags=[FEATURE_TAG],
+            metadata={
+                "thread_id": thread_id,
+                "phase": "resume" if resume else "request",
+            },
+        ),
+    ):
+        try:
+            result = await graph.ainvoke(
+                payload,
+                config={
+                    **config,
+                    "callbacks": [handler],
+                    "run_name": "run-screen-graph",
+                },
+            )
+        except CancelledError:
+            root.update(output={"outcome": "cancelled"})
+            raise
+        except Exception as error:
+            root.update(
+                output={"outcome": "failed", "error_type": type(error).__name__},
+                level="ERROR",
+                status_message=str(error),
+            )
+            raise
+        output = _run_output(result)
+        root.update(
+            output=output,
+            level="ERROR" if output["outcome"] == "failed" else "DEFAULT",
+        )
+        return result
+
+
+def _run_output(state: dict[str, Any]) -> dict[str, Any]:
+    """记录本轮结束原因；completed 表示图已返回，不代表前端应用了动作。"""
+    pending = state.get("__interrupt__")
+    outcome = "completed"
+    if pending:
+        outcome = "waiting_for_user"
+    elif state.get("status") in {"failed", "cancelled"}:
+        outcome = state["status"]
+    return {
+        "outcome": outcome,
+        "intent": state.get("intent"),
+        "answer": state.get("answer"),
+        "clarification": [item.value for item in pending] if pending else None,
+        "actions": state.get("actions"),
+        "validation_passed": state.get("validation_passed"),
+        "failure_reason": state.get("failure_reason"),
+    }
+
+
+@contextmanager
+def observe_decision(request: Any) -> Iterator[Any]:
+    """在当前节点下面记一次 Jev 调用，yield 的是 generation，未追踪时是 None
+
+    TypeSafe 不走 LangChain callback，所以要手记。类型必须是 generation，面板才能算模型和 token
+    只有图回调已经挂上当前 span 时才记。没有父 span 就跳过，避免 Jev 单独变成一条 trace
+    这里不吞异常：失败时标 ERROR 后原样抛出，调用方的 DecisionError 类型不变
+    """
+    if not tracing_enabled() or _ensure_client() is None or not _has_active_span():
+        yield None
+        return
+
+    from langfuse import get_client
+
+    model = request.model or settings.typesafe_model
+    # 意图判断用固定名字，方便以后按名字做评测。别的决策题不要都叫 classify-intent
+    name = "classify-intent" if set(request.decisions) == {"intent"} else "run-decision"
+    with get_client().start_as_current_observation(
+        as_type="generation",
+        name=name,
+        model=model,
+        input={
+            "questions": {
+                name: question.model_dump(mode="json")
+                for name, question in request.decisions.items()
+            },
+            "state": request.state,
+        },
+    ) as generation:
+        try:
+            yield generation
+        except Exception as error:
+            generation.update(
+                level="ERROR",
+                status_message=str(error),
+                metadata=_error_metadata(error),
+            )
+            raise
+
+
+def record_decision(generation: Any, response: Any) -> None:
+    """把成功的 Jev 结果写回刚才那条 generation
+
+    generation 为 None 表示这次没追踪，直接返回，不影响判断结果
+    只写选项、置信度和 token。不写原始 HTTP 响应
+    """
+    if generation is None:
+        return
+    usage: dict[str, int] = {}
+    if response.usage.input_tokens is not None:
+        usage["input_tokens"] = response.usage.input_tokens
+    if response.usage.output_tokens is not None:
+        usage["output_tokens"] = response.usage.output_tokens
+
+    generation.update(
+        model=response.model,
+        output={
+            name: _result_summary(result) for name, result in response.results.items()
+        },
+        usage_details=usage or None,
+        metadata={
+            "request_id": response.request_id or "",
+        },
+    )
+
+
+def _ensure_client() -> Any:
+    """第一次使用时建客户端
+
+    必须在 load_dotenv 之后调用。初始化失败就记一次警告，本进程后续不再重试
+    不要在业务失败时调用，追踪不可用不能挡住图执行
+    """
+    global _client, _unavailable
+
+    if _unavailable:
+        return None
+    if _client is not None:
+        return _client
+    os.environ.setdefault("LANGFUSE_TRACING_ENVIRONMENT", "development")
+    from langfuse import Langfuse
+
+    try:
+        client = Langfuse()
+    except Exception:
+        _unavailable = True
+        logger.warning("Langfuse 不可用，本进程跳过追踪", exc_info=True)
+        return None
+    _client = client
+    return client
+
+
+def _has_active_span() -> bool:
+    """当前调用栈上有没有 Langfuse 已经打开的 span
+
+    图回调会在节点执行期间挂上 span。没有的话，Jev 不该自己开一条 trace
+    """
+    from opentelemetry import trace
+
+    return trace.get_current_span().get_span_context().is_valid
+
+
+def _result_summary(result: Any) -> dict[str, Any]:
+    """按决策题类型抽出面板上要看的字段，不回写原始协议对象"""
+    summary: dict[str, Any] = {"type": result.type}
+
+    if result.type == "choice":
+        summary["choice"] = result.choice
+        summary["confidence"] = result.confidence
+        summary["probabilities"] = result.probabilities
+    elif result.type == "score":
+        summary["score"] = result.score
+        summary["confidence"] = result.confidence
+    elif result.type == "noul":
+        summary["noul"] = result.noul
+    return summary
+
+
+def _error_metadata(error: BaseException) -> dict[str, str]:
+    """失败时附上错误种类和 request_id，方便和 Jev 服务端对日志
+
+    只取已经整理过的安全字段，不把请求体或密钥写进 metadata
+    """
+    metadata = {"error_type": type(error).__name__}
+    kind = getattr(error, "kind", None)
+    request_id = getattr(error, "request_id", None)
+    status = getattr(error, "status", None)
+
+    if isinstance(kind, str):
+        metadata["kind"] = kind
+    if isinstance(request_id, str):
+        metadata["request_id"] = request_id
+    if isinstance(status, int):
+        metadata["status"] = str(status)
+    return metadata

app/processor/graph.py

@@ -23,20 +23,25 @@ from app.processor.state import QueryState
 
 
 def route_after_intent(state: QueryState) -> str:
+    # 按最终意图分流：问答进入 QA，意图不明进入澄清，新建或修改进入需求理解
     return state["intent"]
 
 
 def route_after_requirement(state: QueryState) -> bool:
+    # True 进入澄清，False 进入修改范围判断
     return state["requirement"].needs_clarification
 
 
 def route_after_wait(state: QueryState) -> str:
+    # 用户补充后回到输入节点合并上下文，取消则结束流程
     return "input_node" if state["status"] == "running" else END
 
 
 def route_after_scope(state: QueryState) -> str:
+    # 需要规划时先获取组件能力，供规划节点选择可用组件
     if state["change_scope"].needs_planning:
         return "component_resource_node"
+    # 已支持的简单修改直接生成动作，尚不支持的需求暂时结束流程
     return "action_node" if can_generate_actions(state) else END
 
 
@@ -208,6 +213,7 @@ if __name__ == "__main__":
     from pydantic import BaseModel
 
     from app.clients.decision import close_decision_engine
+    from app.observability.tracing import ainvoke_screen, flush_tracing
 
     async def main():
         # 1. 修改请求，测试完整流程
@@ -230,10 +236,17 @@ if __name__ == "__main__":
                 "context": {"selected_node_ids": ["sales-chart"]},
             }
         }
+
         # 2. 执行图，程序退出时关闭共享连接
-        config = {"configurable": {"thread_id": str(uuid4())}}
+        thread_id = str(uuid4())
+        session_id = (
+            state["request"].get("context", {}).get("conversation_id") or thread_id
+        )
+
         try:
-            final_state = await query_app.ainvoke(state, config=config)
+            final_state = await ainvoke_screen(
+                query_app, state, thread_id=thread_id, session_id=session_id
+            )
             while final_state.get("__interrupt__"):
                 pending = final_state["__interrupt__"][0].value
                 print(pending["question"])
@@ -252,11 +265,15 @@ if __name__ == "__main__":
                     notes = input("补充说明（可留空）：").strip()
                     if notes:
                         reply["answer"] = notes
-                final_state = await query_app.ainvoke(
+                final_state = await ainvoke_screen(
-                    Command(resume=reply), config=config
+                    query_app,
+                    Command(resume=reply),
+                    thread_id=thread_id,
+                    session_id=session_id,
                 )
         finally:
             await close_decision_engine()
+            flush_tracing()
 
         # 3. 打印完整 state
         print(

app/processor/nodes/clarify_node.py

@@ -46,6 +46,7 @@ if __name__ == "__main__":
     from pydantic import BaseModel
 
     from app.clients.decision import close_decision_engine
+    from app.observability.tracing import ainvoke_screen, flush_tracing
     from app.processor.graph import query_app
 
     async def main():
@@ -57,11 +58,12 @@ if __name__ == "__main__":
             }
         }
         try:
-            final_state = await query_app.ainvoke(
+            final_state = await ainvoke_screen(
-                state, config={"configurable": {"thread_id": "clarify-test"}}
+                query_app, state, thread_id="clarify-test"
             )
         finally:
             await close_decision_engine()
+            flush_tracing()
         print(
             json.dumps(
                 {