from __future__ import annotations import asyncio from typing import Any import httpx from fastapi import HTTPException, status from aiprovider.config import settings from aiprovider.schemas import ( AIProviderStatusResponse, SituationalAnalysisRequest, SituationalAnalysisResponse, ) def _normalize_provider(value: str) -> str: return (value or "disabled").strip().lower() class ProviderService: def __init__(self) -> None: self.provider = _normalize_provider(settings.AI_PROVIDER) self.base_url = settings.AI_BASE_URL.rstrip("/") self.api_key = settings.AI_API_KEY self.default_model = settings.AI_MODEL self.timeout = settings.AI_TIMEOUT_SECONDS self.http_retry_attempts = max(settings.AI_HTTP_RETRY_ATTEMPTS, 1) self.max_tokens = settings.AI_MAX_TOKENS self.anthropic_version = settings.AI_ANTHROPIC_VERSION self.system_prompt = settings.AI_ANALYSIS_SYSTEM_PROMPT def get_status(self) -> AIProviderStatusResponse: enabled = self.provider != "disabled" configured = enabled and bool(self.base_url and self.api_key and self.default_model) return AIProviderStatusResponse( provider=self.provider, enabled=enabled, configured=configured, model=self.default_model or None, base_url=self.base_url if enabled else None, ) async def analyze(self, payload: SituationalAnalysisRequest) -> SituationalAnalysisResponse: if self.provider == "disabled": raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="AI provider is disabled. Configure AI_PROVIDER in .env to enable analysis.", ) model = payload.preferred_model or self.default_model if not self.base_url or not self.api_key or not model: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="AI provider is not fully configured. Check AI_BASE_URL, AI_API_KEY, and AI_MODEL.", ) prompt = self._build_prompt(payload) if self.provider in {"openai", "openai_compatible"}: data = await self._request_openai_compatible(model, prompt) content = self._extract_openai_content(data) elif self.provider in {"anthropic", "anthropic_compatible", "claude_compatible"}: data = await self._request_anthropic_compatible(model, prompt) content = self._extract_anthropic_content(data) elif self.provider == "ollama": data = await self._request_ollama(model, prompt) content = self._extract_ollama_content(data) else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported AI provider: {self.provider}", ) return SituationalAnalysisResponse( provider=self.provider, model=model, content=content, raw_response=data, ) def _build_prompt(self, payload: SituationalAnalysisRequest) -> str: sections = [ f"任务标题:\n{payload.title}", f"分析目标:\n{payload.objective}", ] if payload.observations: sections.append("观测事实:\n" + "\n".join(f"- {item}" for item in payload.observations)) if payload.constraints: sections.append("约束条件:\n" + "\n".join(f"- {item}" for item in payload.constraints)) if payload.context: sections.append(f"附加上下文:\n{payload.context}") sections.append( "请输出: 1) 态势摘要 2) 关键风险 3) 研判依据 4) 建议动作 5) 还缺少的数据。" ) return "\n\n".join(sections) async def _request_openai_compatible(self, model: str, prompt: str) -> dict[str, Any]: request_body = { "model": model, "messages": [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": prompt}, ], "temperature": 0.2, } return await self._post( path="/chat/completions", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, request_body=request_body, ) async def _request_anthropic_compatible(self, model: str, prompt: str) -> dict[str, Any]: request_body = { "model": model, "system": self.system_prompt, "messages": [ { "role": "user", "content": [ { "type": "text", "text": prompt, } ], } ], "max_tokens": self.max_tokens, "temperature": 0.2, } return await self._post( path="/messages", headers={ "x-api-key": self.api_key, "anthropic-version": self.anthropic_version, "Content-Type": "application/json", }, request_body=request_body, ) async def _request_ollama(self, model: str, prompt: str) -> dict[str, Any]: request_body = { "model": model, "stream": False, "system": self.system_prompt, "prompt": prompt, "options": { "temperature": 0.2, }, } return await self._post( path="/api/generate", headers={ "Content-Type": "application/json", }, request_body=request_body, ) async def _post( self, path: str, headers: dict[str, str], request_body: dict[str, Any], ) -> dict[str, Any]: last_error: Exception | None = None for attempt in range(1, self.http_retry_attempts + 1): try: async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.post( f"{self.base_url}{path}", headers=headers, json=request_body, ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as exc: last_error = exc if attempt < self.http_retry_attempts and exc.response.status_code >= 500: await asyncio.sleep(0.3 * attempt) continue detail = exc.response.text or "AI provider returned an error" raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"AI provider request failed: {detail}", ) from exc except httpx.HTTPError as exc: last_error = exc if attempt < self.http_retry_attempts: await asyncio.sleep(0.3 * attempt) continue raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Failed to reach AI provider: {exc}", ) from exc raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"AI provider request failed: {last_error}", ) def _extract_openai_content(self, payload: dict[str, Any]) -> str: choices = payload.get("choices") or [] if not choices: return "" message = choices[0].get("message") or {} content = message.get("content") if isinstance(content, str): return content if isinstance(content, list): return "".join( item.get("text", "") for item in content if isinstance(item, dict) ) return "" def _extract_anthropic_content(self, payload: dict[str, Any]) -> str: content = payload.get("content") if isinstance(content, str): return content if not isinstance(content, list): return "" fragments: list[str] = [] for item in content: if not isinstance(item, dict): continue if item.get("type") == "text" and isinstance(item.get("text"), str): fragments.append(item["text"]) return "".join(fragments) def _extract_ollama_content(self, payload: dict[str, Any]) -> str: response = payload.get("response") if isinstance(response, str): return response return ""