cfc/cfc-langgraph/src/app.py

import json
from datetime import datetime

from dotenv import load_dotenv
load_dotenv()

from fastapi import APIRouter, FastAPI
from fastapi.responses import JSONResponse

from .schemas.questionnaire import GenerateRequest, GenerateResponse
from .graphs.questionnaire import get_questionnaire_graph
from .schemas.emotion import EmotionRequest, EmotionResponse, EmotionItem
from .graphs.emotion import get_emotion_graph

router = APIRouter(prefix="/api/v1", tags=["questionnaire"])


@router.post("/questionnaire/generate")
async def generate_questionnaire(req: GenerateRequest):
    graph = get_questionnaire_graph()
    try:
        result = graph.invoke({
            "request": req,
            "raw_response": "",
            "questionnaire": None,
            "error": None,
        })
        if result.get("error"):
            return JSONResponse(
                status_code=500,
                content={"error": result["error"]}
            )
        version = datetime.now().strftime("%Y%m%d%H%M%S")
        return GenerateResponse(
            questionnaire_json=json.dumps(result["questionnaire"], ensure_ascii=False),
            version=version,
        )
    except Exception as e:
        return JSONResponse(
            status_code=500,
            content={"error": f"graph 执行失败: {str(e)}"}
        )


app = FastAPI(title="CFC LangGraph 问卷生成服务")
app.include_router(router)


@app.get("/health")
def health():
    return {"status": "ok"}


# ── 情绪识别 ─────────────────────────────────────────────────

@router.post("/emotion/recognize")
async def recognize_emotion(req: EmotionRequest):
    graph = get_emotion_graph()
    try:
        result = graph.invoke({
            "request": req.model_dump(),
            "image_bytes": None,
            "image_path": None,
            "emotion_result": None,
            "emotions": [],
            "dominant_emotion": "neutral",
            "dominant_label_zh": "平静",
            "all_emotions": {},
            "error": None,
        })
        if result.get("error"):
            return JSONResponse(
                status_code=400,
                content={"error": result["error"]}
            )
        return EmotionResponse(
            emotions=[
                EmotionItem(emotion=e["emotion"], confidence=e["confidence"])
                for e in result["emotions"]
            ],
            dominant_emotion=result["dominant_emotion"],
            dominant_label_zh=result["dominant_label_zh"],
            all_emotions=result["all_emotions"],
        )
    except Exception as e:
        return JSONResponse(
            status_code=500,
            content={"error": f"graph 执行失败: {str(e)}"}
        )