cfc/cfc-langgraph/app/monitoring.py

"""Prometheus 监控指标"""
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import APIRouter, Response
import time
import functools


llm_calls_total = Counter(
    "llm_calls_total", "Total LLM API calls",
    ["model", "status"],
)
llm_duration_seconds = Histogram(
    "llm_duration_seconds", "LLM call duration",
    ["model"],
    buckets=(0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0),
)
llm_tokens_total = Counter(
    "llm_tokens_total", "Total tokens used",
    ["model", "type"],
)

rag_retrievals_total = Counter(
    "rag_retrievals_total", "Total RAG retrievals",
    ["method"],
)
rag_duration_seconds = Histogram(
    "rag_duration_seconds", "RAG retrieval duration",
    ["method"],
    buckets=(0.01, 0.05, 0.1, 0.5, 1.0),
)

agent_calls_total = Counter(
    "agent_calls_total", "Total Agent invocations",
    ["agent_type"],
)
agent_duration_seconds = Histogram(
    "agent_duration_seconds", "Agent execution duration",
    ["agent_type"],
    buckets=(0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0),
)

java_calls_total = Counter(
    "java_calls_total", "Total calls to Java backend",
    ["endpoint", "status"],
)
java_duration_seconds = Histogram(
    "java_duration_seconds", "Java backend call duration",
    ["endpoint"],
    buckets=(0.01, 0.05, 0.1, 0.5, 1.0, 2.0),
)

kb_sync_duration = Gauge(
    "kb_sync_duration_seconds", "Last knowledge base sync duration"
)
kb_sync_documents = Gauge(
    "kb_sync_documents_total", "Documents processed in last sync"
)


def monitor_agent(agent_type: str):
    """Agent 性能监控装饰器"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            agent_calls_total.labels(agent_type=agent_type).inc()
            start = time.perf_counter()
            try:
                result = await func(*args, **kwargs)
                agent_duration_seconds.labels(agent_type=agent_type).observe(
                    time.perf_counter() - start)
                return result
            except Exception as e:
                agent_duration_seconds.labels(agent_type=agent_type).observe(
                    time.perf_counter() - start)
                raise
        return wrapper
    return decorator


router = APIRouter(tags=["monitoring"])


@router.get("/metrics")
async def metrics():
    return Response(
        content=generate_latest(),
        media_type="text/plain; charset=utf-8",
    )


@router.get("/api/v1/health")
async def detailed_health():
    """详细健康检查 (含组件状态)"""
    from app.config import settings
    status = {"status": "ok", "components": {}}

    try:
        import os
        chroma_path = settings.chroma_db_path
        status["components"]["chromadb"] = {
            "status": "ok",
            "path": chroma_path,
            "exists": os.path.exists(chroma_path),
        }
    except Exception as e:
        status["components"]["chromadb"] = {"status": "error", "message": str(e)}
        status["status"] = "degraded"

    try:
        from langchain_openai import ChatOpenAI
        llm = ChatOpenAI(
            model=settings.llm_model,
            api_key=settings.llm_api_key,
            base_url=settings.llm_base_url,
            max_tokens=5,
        )
        await llm.ainvoke("ping")
        status["components"]["llm"] = {"status": "ok"}
    except Exception as e:
        status["components"]["llm"] = {"status": "error", "message": str(e)}
        status["status"] = "degraded"

    try:
        from app.tools.java_client import JavaClient
        client = JavaClient()
        jc = await client._get_client()
        resp = await jc.get("/health")
        if resp.status_code == 200:
            status["components"]["java_backend"] = {"status": "ok"}
        else:
            status["components"]["java_backend"] = {"status": "error", "code": resp.status_code}
            status["status"] = "degraded"
    except Exception as e:
        status["components"]["java_backend"] = {"status": "error", "message": str(e)}
        status["status"] = "degraded"

    return status