feat(api): Phase 15.1 Langfuse LLMOps 整合 + 模型升級
## 新功能 - Langfuse 自建部署 (192.168.0.110:3100) - langfuse_client.py - LLM 呼叫追蹤包裝 - OpenClaw 整合 Langfuse trace ## 模型升級 (統帥批准) - 生產預設: llama3.2:3b → qwen2.5:7b-instruct - 摘要任務: llama3.2:3b (速度優先) ## 配置更新 - requirements.txt: +langfuse>=2.0.0 - config.py: +LANGFUSE_* 設定 - models.json: 更新 Ollama 模型配置 - K8s: Secret + ConfigMap 更新 ## 審查通過 - 模組化檢查 ✅ - 核心測試 31/31 ✅ Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
285
apps/api/src/services/langfuse_client.py
Normal file
285
apps/api/src/services/langfuse_client.py
Normal file
@@ -0,0 +1,285 @@
|
||||
"""
|
||||
Langfuse LLMOps Client - Phase 15.1
|
||||
===================================
|
||||
LLM 呼叫追蹤、成本監控、Prompt 版本管理
|
||||
|
||||
Phase 15.1 (2026-03-26)
|
||||
端點: http://192.168.0.110:3100 (DevOps 金庫)
|
||||
|
||||
Features:
|
||||
- 自動追蹤所有 LLM 呼叫 (Ollama/Gemini/Claude)
|
||||
- 成本估算與監控
|
||||
- Prompt 版本管理
|
||||
- 與 OTEL Trace 整合
|
||||
|
||||
Usage:
|
||||
from src.services.langfuse_client import get_langfuse, langfuse_trace
|
||||
|
||||
# 方法 1: Context Manager
|
||||
async with langfuse_trace("openclaw_decision") as trace:
|
||||
result = await call_llm(prompt)
|
||||
trace.generation(
|
||||
name="ollama_call",
|
||||
model="qwen2.5:7b-instruct",
|
||||
input=prompt,
|
||||
output=result,
|
||||
)
|
||||
|
||||
# 方法 2: 裝飾器
|
||||
@langfuse_observe(name="analyze_incident")
|
||||
async def analyze_incident(incident_id: str):
|
||||
...
|
||||
"""
|
||||
|
||||
from contextlib import asynccontextmanager
|
||||
from functools import wraps
|
||||
from typing import Any, Callable
|
||||
|
||||
import structlog
|
||||
|
||||
from src.core.config import settings
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
# Langfuse client singleton
|
||||
_langfuse_client = None
|
||||
|
||||
|
||||
def get_langfuse():
|
||||
"""
|
||||
取得 Langfuse client singleton
|
||||
|
||||
Returns:
|
||||
Langfuse client 或 None (如果未啟用或未配置)
|
||||
"""
|
||||
global _langfuse_client
|
||||
|
||||
if not settings.LANGFUSE_ENABLED:
|
||||
return None
|
||||
|
||||
if not settings.LANGFUSE_PUBLIC_KEY or not settings.LANGFUSE_SECRET_KEY:
|
||||
logger.warning(
|
||||
"langfuse_not_configured",
|
||||
message="Langfuse enabled but keys not set",
|
||||
)
|
||||
return None
|
||||
|
||||
if _langfuse_client is None:
|
||||
try:
|
||||
from langfuse import Langfuse
|
||||
|
||||
_langfuse_client = Langfuse(
|
||||
public_key=settings.LANGFUSE_PUBLIC_KEY,
|
||||
secret_key=settings.LANGFUSE_SECRET_KEY,
|
||||
host=settings.LANGFUSE_URL,
|
||||
)
|
||||
logger.info(
|
||||
"langfuse_initialized",
|
||||
host=settings.LANGFUSE_URL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"langfuse_init_failed",
|
||||
error=str(e),
|
||||
)
|
||||
return None
|
||||
|
||||
return _langfuse_client
|
||||
|
||||
|
||||
class LangfuseTraceContext:
|
||||
"""Langfuse Trace Context for tracking LLM calls"""
|
||||
|
||||
def __init__(self, name: str, metadata: dict[str, Any] | None = None):
|
||||
self.name = name
|
||||
self.metadata = metadata or {}
|
||||
self.trace = None
|
||||
self._client = get_langfuse()
|
||||
|
||||
def __enter__(self):
|
||||
if self._client:
|
||||
try:
|
||||
self.trace = self._client.trace(
|
||||
name=self.name,
|
||||
metadata=self.metadata,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning("langfuse_trace_start_failed", error=str(e))
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
# Langfuse auto-flushes, no explicit close needed
|
||||
pass
|
||||
|
||||
def generation(
|
||||
self,
|
||||
name: str,
|
||||
model: str,
|
||||
input: str | dict[str, Any],
|
||||
output: str | dict[str, Any] | None = None,
|
||||
usage: dict[str, int] | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
):
|
||||
"""
|
||||
記錄一次 LLM generation
|
||||
|
||||
Args:
|
||||
name: Generation 名稱 (e.g., "ollama_call", "gemini_fallback")
|
||||
model: 模型名稱 (e.g., "qwen2.5:7b-instruct", "gemini-1.5-flash")
|
||||
input: 輸入 prompt
|
||||
output: 輸出結果
|
||||
usage: Token 使用量 {"input": x, "output": y}
|
||||
metadata: 額外 metadata
|
||||
"""
|
||||
if not self.trace:
|
||||
return None
|
||||
|
||||
try:
|
||||
gen = self.trace.generation(
|
||||
name=name,
|
||||
model=model,
|
||||
input=input,
|
||||
output=output,
|
||||
usage=usage,
|
||||
metadata=metadata or {},
|
||||
)
|
||||
return gen
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"langfuse_generation_failed",
|
||||
error=str(e),
|
||||
name=name,
|
||||
model=model,
|
||||
)
|
||||
return None
|
||||
|
||||
def span(self, name: str, metadata: dict[str, Any] | None = None):
|
||||
"""
|
||||
記錄一個 span (非 LLM 操作)
|
||||
|
||||
Args:
|
||||
name: Span 名稱
|
||||
metadata: 額外 metadata
|
||||
"""
|
||||
if not self.trace:
|
||||
return None
|
||||
|
||||
try:
|
||||
return self.trace.span(name=name, metadata=metadata or {})
|
||||
except Exception as e:
|
||||
logger.warning("langfuse_span_failed", error=str(e), name=name)
|
||||
return None
|
||||
|
||||
def score(
|
||||
self,
|
||||
name: str,
|
||||
value: float,
|
||||
comment: str | None = None,
|
||||
):
|
||||
"""
|
||||
記錄評分 (用於 Prompt 品質追蹤)
|
||||
|
||||
Args:
|
||||
name: 評分名稱 (e.g., "response_quality", "format_compliance")
|
||||
value: 分數 (0.0 - 1.0)
|
||||
comment: 評論
|
||||
"""
|
||||
if not self.trace:
|
||||
return
|
||||
|
||||
try:
|
||||
self.trace.score(
|
||||
name=name,
|
||||
value=value,
|
||||
comment=comment,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"langfuse_score_failed",
|
||||
error=str(e),
|
||||
name=name,
|
||||
)
|
||||
|
||||
|
||||
def langfuse_trace(name: str, metadata: dict[str, Any] | None = None):
|
||||
"""
|
||||
Langfuse trace context manager
|
||||
|
||||
Usage:
|
||||
with langfuse_trace("openclaw_decision") as trace:
|
||||
result = await call_llm(prompt)
|
||||
trace.generation(name="ollama", model="qwen2.5:7b-instruct", ...)
|
||||
"""
|
||||
return LangfuseTraceContext(name=name, metadata=metadata)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def langfuse_trace_async(name: str, metadata: dict[str, Any] | None = None):
|
||||
"""
|
||||
Async version of langfuse_trace
|
||||
|
||||
Usage:
|
||||
async with langfuse_trace_async("openclaw_decision") as trace:
|
||||
result = await call_llm(prompt)
|
||||
"""
|
||||
ctx = LangfuseTraceContext(name=name, metadata=metadata)
|
||||
ctx.__enter__()
|
||||
try:
|
||||
yield ctx
|
||||
finally:
|
||||
ctx.__exit__(None, None, None)
|
||||
|
||||
|
||||
def langfuse_observe(
|
||||
name: str | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
):
|
||||
"""
|
||||
Langfuse 裝飾器 - 自動追蹤函數執行
|
||||
|
||||
Usage:
|
||||
@langfuse_observe(name="analyze_incident")
|
||||
async def analyze_incident(incident_id: str):
|
||||
...
|
||||
"""
|
||||
|
||||
def decorator(func: Callable):
|
||||
trace_name = name or func.__name__
|
||||
|
||||
@wraps(func)
|
||||
async def async_wrapper(*args, **kwargs):
|
||||
async with langfuse_trace_async(trace_name, metadata) as trace:
|
||||
# Inject trace into kwargs if function accepts it
|
||||
if "langfuse_trace" in func.__code__.co_varnames:
|
||||
kwargs["langfuse_trace"] = trace
|
||||
return await func(*args, **kwargs)
|
||||
|
||||
@wraps(func)
|
||||
def sync_wrapper(*args, **kwargs):
|
||||
with langfuse_trace(trace_name, metadata) as trace:
|
||||
if "langfuse_trace" in func.__code__.co_varnames:
|
||||
kwargs["langfuse_trace"] = trace
|
||||
return func(*args, **kwargs)
|
||||
|
||||
# Return appropriate wrapper based on function type
|
||||
import asyncio
|
||||
|
||||
if asyncio.iscoroutinefunction(func):
|
||||
return async_wrapper
|
||||
return sync_wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
def flush_langfuse():
|
||||
"""
|
||||
手動 flush Langfuse (通常不需要，client 會自動 flush)
|
||||
用於測試或確保資料送出
|
||||
"""
|
||||
client = get_langfuse()
|
||||
if client:
|
||||
try:
|
||||
client.flush()
|
||||
logger.debug("langfuse_flushed")
|
||||
except Exception as e:
|
||||
logger.warning("langfuse_flush_failed", error=str(e))
|
||||
@@ -33,6 +33,7 @@ from src.core.redis_client import get_redis
|
||||
from src.models.ai import (
|
||||
OpenClawDecision,
|
||||
)
|
||||
from src.services.langfuse_client import langfuse_trace
|
||||
from src.services.signoz_client import GoldMetrics, get_signoz_client
|
||||
from src.utils.timezone import now_taipei_iso
|
||||
|
||||
@@ -360,7 +361,7 @@ class OpenClawService:
|
||||
response = await client.post(
|
||||
f"{settings.OLLAMA_URL}/api/generate",
|
||||
json={
|
||||
"model": "llama3.2:3b", # 使用更大的模型提高品質
|
||||
"model": "qwen2.5:7b-instruct", # 使用更大的模型提高品質
|
||||
"prompt": prompt,
|
||||
"stream": False,
|
||||
"format": "json", # 強制 JSON 輸出
|
||||
@@ -823,34 +824,75 @@ class OpenClawService:
|
||||
|
||||
若 MOCK_MODE=True，直接回傳模擬結果。
|
||||
若所有 Provider 失敗，fallback 到 Mock。
|
||||
|
||||
Phase 15.1: 整合 Langfuse LLMOps 追蹤
|
||||
"""
|
||||
# Mock Mode: 開發測試用
|
||||
if settings.MOCK_MODE:
|
||||
logger.info("mock_mode_enabled", using="mock_llm")
|
||||
return self._generate_mock_response(alert_context or {}, signoz_metrics), "mock", True
|
||||
|
||||
for provider in settings.AI_FALLBACK_ORDER:
|
||||
logger.info("ai_provider_attempt", provider=provider)
|
||||
# Phase 15.1: Langfuse 追蹤整合
|
||||
with langfuse_trace(
|
||||
"openclaw_fallback_chain",
|
||||
metadata={
|
||||
"prompt_length": len(prompt),
|
||||
"fallback_order": settings.AI_FALLBACK_ORDER,
|
||||
"alert_fingerprint": (alert_context or {}).get("fingerprint", "unknown"),
|
||||
},
|
||||
) as trace:
|
||||
for provider in settings.AI_FALLBACK_ORDER:
|
||||
logger.info("ai_provider_attempt", provider=provider)
|
||||
|
||||
if provider == "ollama":
|
||||
response, success = await self._call_ollama(prompt)
|
||||
elif provider == "gemini":
|
||||
response, success = await self._call_gemini(prompt)
|
||||
elif provider == "claude":
|
||||
response, success = await self._call_claude(prompt)
|
||||
else:
|
||||
logger.warning("unknown_ai_provider", provider=provider)
|
||||
continue
|
||||
start_time = time.time()
|
||||
model_name = self._get_model_name(provider)
|
||||
|
||||
if success:
|
||||
logger.info("ai_provider_success", provider=provider)
|
||||
return response, provider, True
|
||||
if provider == "ollama":
|
||||
response, success = await self._call_ollama(prompt)
|
||||
elif provider == "gemini":
|
||||
response, success = await self._call_gemini(prompt)
|
||||
elif provider == "claude":
|
||||
response, success = await self._call_claude(prompt)
|
||||
else:
|
||||
logger.warning("unknown_ai_provider", provider=provider)
|
||||
continue
|
||||
|
||||
logger.warning("ai_provider_failed_fallback", provider=provider)
|
||||
latency_ms = (time.time() - start_time) * 1000
|
||||
|
||||
# 所有 Provider 失敗時，fallback 到 Mock (優雅降級)
|
||||
logger.warning("all_providers_failed_using_mock", fallback="mock_llm")
|
||||
return self._generate_mock_response(alert_context or {}, signoz_metrics), "mock_fallback", True
|
||||
# Langfuse: 記錄每次 LLM 呼叫
|
||||
trace.generation(
|
||||
name=f"{provider}_call",
|
||||
model=model_name,
|
||||
input=prompt[:500], # 截斷避免過長
|
||||
output=response[:500] if success else f"ERROR: {response[:200]}",
|
||||
metadata={
|
||||
"success": success,
|
||||
"latency_ms": round(latency_ms, 2),
|
||||
"provider": provider,
|
||||
},
|
||||
)
|
||||
|
||||
if success:
|
||||
logger.info("ai_provider_success", provider=provider, latency_ms=latency_ms)
|
||||
# Langfuse: 記錄成功評分
|
||||
trace.score(name="provider_success", value=1.0, comment=f"Success via {provider}")
|
||||
return response, provider, True
|
||||
|
||||
logger.warning("ai_provider_failed_fallback", provider=provider, latency_ms=latency_ms)
|
||||
|
||||
# 所有 Provider 失敗時，fallback 到 Mock (優雅降級)
|
||||
logger.warning("all_providers_failed_using_mock", fallback="mock_llm")
|
||||
trace.score(name="provider_success", value=0.0, comment="All providers failed, using mock")
|
||||
return self._generate_mock_response(alert_context or {}, signoz_metrics), "mock_fallback", True
|
||||
|
||||
def _get_model_name(self, provider: str) -> str:
|
||||
"""取得 provider 對應的模型名稱"""
|
||||
model_map = {
|
||||
"ollama": "qwen2.5:7b-instruct",
|
||||
"gemini": "gemini-1.5-flash",
|
||||
"claude": "claude-3-haiku-20240307",
|
||||
}
|
||||
return model_map.get(provider, provider)
|
||||
|
||||
# =========================================================================
|
||||
# Response Parsing (防禦性解析)
|
||||
|
||||
Reference in New Issue
Block a user