diff --git a/.env.example b/.env.example index 1c45128..29e7f0d 100644 --- a/.env.example +++ b/.env.example @@ -14,4 +14,13 @@ TYPESAFE_MAX_RETRIES=2 INTENT_MIN_CONFIDENCE=0.7 # 动作自动修正次数,不含首次生成,不限制用户澄清轮数 -MAX_ACTION_RETRIES=2 \ No newline at end of file +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 diff --git a/app/clients/decision.py b/app/clients/decision.py index e0569e1..95c5aa8 100644 --- a/app/clients/decision.py +++ b/app/clients/decision.py @@ -77,34 +77,41 @@ class DecisionEngine: | {"type": question.type} for name, question in request.decisions.items() } - try: - response = await client.system_one( - state=request.state, - questions=questions, - model=request.model, - extra_body=request.extra_body or None, - ) - except TypeSafeAPIResponseValidationError as error: - raise DecisionError( - "Jev 返回的数据格式无效", - kind="response_validation", - status=error.status, - request_id=error.request_id, - ) from error - except TypeSafeAPIError as error: - raise DecisionError( - f"Jev 请求失败(HTTP {error.status})", - kind="api", - status=error.status, - request_id=error.request_id, - ) from error - except TypeSafeAPITimeoutError as error: - raise DecisionError("Jev 请求超时", kind="timeout") from error - except TypeSafeAPIConnectionError as error: - 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) + 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, + questions=questions, + model=request.model, + extra_body=request.extra_body or None, + ) + except TypeSafeAPIResponseValidationError as error: + raise DecisionError( + "Jev 返回的数据格式无效", + kind="response_validation", + status=error.status, + request_id=error.request_id, + ) from error + except TypeSafeAPIError as error: + raise DecisionError( + f"Jev 请求失败(HTTP {error.status})", + kind="api", + status=error.status, + request_id=error.request_id, + ) from error + except TypeSafeAPITimeoutError as error: + raise DecisionError("Jev 请求超时", kind="timeout") from error + except TypeSafeAPIConnectionError as error: + raise DecisionError("无法连接 Jev 服务", kind="connection") from error + except TypeSafeError as error: + raise DecisionError("Jev SDK 调用失败", kind="sdk") from error + parsed = self._parse_response(request, response) + record_decision(generation, parsed) + return parsed @staticmethod def _parse_response( diff --git a/app/observability/__init__.py b/app/observability/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/observability/tracing.py b/app/observability/tracing.py new file mode 100644 index 0000000..3886724 --- /dev/null +++ b/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 diff --git a/app/processor/graph.py b/app/processor/graph.py index 330fcb4..d07ab79 100644 --- a/app/processor/graph.py +++ b/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( - Command(resume=reply), config=config + final_state = await ainvoke_screen( + query_app, + Command(resume=reply), + thread_id=thread_id, + session_id=session_id, ) finally: await close_decision_engine() + flush_tracing() # 3. 打印完整 state print( diff --git a/app/processor/nodes/clarify_node.py b/app/processor/nodes/clarify_node.py index e5e317a..21f61f6 100644 --- a/app/processor/nodes/clarify_node.py +++ b/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( - state, config={"configurable": {"thread_id": "clarify-test"}} + final_state = await ainvoke_screen( + query_app, state, thread_id="clarify-test" ) finally: await close_decision_engine() + flush_tracing() print( json.dumps( {