import httpx from typing import Optional from app.config import settings import logging logger = logging.getLogger(__name__) class TongueDiagnosisAgent: """舌诊分析 Agent 当前实现: 代理到 Dify Workflow (多模态最成熟) 后续可替换: 直接调用多模态 LLM API """ def __init__(self): self.dify_base = settings.dify_base_url or "" self.dify_api_key = settings.dify_tongue_api_key or "" async def diagnose( self, image_url: str, user_id: int, additional_context: Optional[dict] = None, ) -> dict: """舌诊分析: 调用 Dify Workflow 或直接 LLM""" if self.dify_base and self.dify_api_key: return await self._via_dify(image_url, user_id, additional_context) else: return await self._via_llm(image_url) async def _via_dify( self, image_url: str, user_id: int, context: Optional[dict] ) -> dict: """通过 Dify Workflow 执行舌诊""" url = f"{self.dify_base}/workflows/run" headers = { "Authorization": f"Bearer {self.dify_api_key}", "Content-Type": "application/json", } inputs = {"tongue_image": {"type": "image", "url": image_url}} if context: inputs.update(context) body = { "inputs": inputs, "user": str(user_id), "response_mode": "blocking", } try: async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(url, json=body, headers=headers) data = resp.json() if "data" in data and "outputs" in data["data"]: return data["data"]["outputs"] except Exception as e: logger.warning("Dify 舌诊失败: %s", e) return self._mock_result() async def _via_llm(self, image_url: str) -> dict: """直接调用多模态 LLM (预留)""" logger.warning("多模态 LLM 未配置, 返回模拟数据") return self._mock_result() def _mock_result(self) -> dict: return { "overall_assessment": "舌象基本正常, 舌质淡红, 苔薄白, 提示脾胃功能尚可。", "indicators": [ {"code": "tongue_color", "value": "淡红"}, {"code": "coating_color", "value": "薄白"}, {"code": "coating_texture", "value": "润"}, {"code": "fissure", "value": "无"}, {"code": "teeth_mark", "value": "轻"}, {"code": "sublingual_vein", "value": "正常"}, {"code": "constitution", "value": "平和质"}, ], }