planet/aiprovider/provider_service.py

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 (
    AIContentBlock,
    AIProviderStatusResponse,
    SituationalAnalysisRequest,
    SituationalAnalysisResponse,
)


def _normalize_provider(value: str) -> str:
    return (value or "disabled").strip().lower()


def _normalize_provider_api(value: str) -> str:
    return (value or "auto").strip().lower().replace("_", "-")


def _resolve_provider_api(provider: str, configured_api: str) -> str:
    if configured_api and configured_api != "auto":
        return configured_api

    if provider in {"openai", "openai-compatible", "openai_compatible"}:
        return "openai-completions"
    if provider in {
        "anthropic",
        "anthropic-compatible",
        "anthropic_compatible",
        "claude-compatible",
        "claude_compatible",
        "minimax",
        "kimi-coding",
        "moonshot-anthropic",
    }:
        return "anthropic-messages"
    if provider == "ollama":
        return "ollama-generate"
    return "disabled"


class ProviderService:
    def __init__(self, overrides: dict[str, Any] | None = None) -> None:
        overrides = overrides or {}
        self.provider = _normalize_provider(overrides.get("provider") or settings.AI_PROVIDER)
        self.provider_api = _resolve_provider_api(
            self.provider,
            _normalize_provider_api(overrides.get("provider_api") or settings.AI_PROVIDER_API),
        )
        self.base_url = str(overrides.get("base_url") or settings.AI_BASE_URL).rstrip("/")
        self.api_key = str(overrides.get("api_key") or settings.AI_API_KEY)
        self.default_model = str(overrides.get("model") or settings.AI_MODEL)
        self.timeout = settings.AI_TIMEOUT_SECONDS
        self.http_retry_attempts = max(settings.AI_HTTP_RETRY_ATTEMPTS, 1)
        self.max_tokens = int(overrides.get("max_tokens") or settings.AI_MAX_TOKENS)
        self.anthropic_version = str(
            overrides.get("anthropic_version") or settings.AI_ANTHROPIC_VERSION
        )
        self.system_prompt = settings.AI_ANALYSIS_SYSTEM_PROMPT

    def get_status(self) -> AIProviderStatusResponse:
        enabled = self.provider != "disabled"
        has_credentials = bool(self.api_key) if self._requires_api_key() else True
        configured = enabled and bool(self.base_url and has_credentials and self.default_model)
        return AIProviderStatusResponse(
            provider=self.provider,
            api=self.provider_api if enabled else None,
            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
        has_credentials = bool(self.api_key) if self._requires_api_key() else True
        if not self.base_url or not has_credentials 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_api == "openai-completions":
            data = await self._request_openai_compatible(model, prompt)
            content = self._extract_openai_content(data)
            content_blocks = self._extract_openai_blocks(data)
        elif self.provider_api == "anthropic-messages":
            data = await self._request_anthropic_messages(model, prompt, payload.thinking)
            content = self._extract_anthropic_content(data)
            content_blocks = self._extract_anthropic_blocks(data)
        elif self.provider_api == "ollama-generate":
            data = await self._request_ollama(model, prompt)
            content = self._extract_ollama_content(data)
            content_blocks = self._extract_ollama_blocks(data)
        else:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Unsupported AI provider API: {self.provider_api}",
            )

        text_blocks = [block.text for block in content_blocks if block.text]
        thinking_blocks = [block.thinking for block in content_blocks if block.thinking]

        return SituationalAnalysisResponse(
            provider=self.provider,
            model=model,
            content=content,
            content_blocks=content_blocks,
            text_blocks=text_blocks,
            thinking_blocks=thinking_blocks,
            raw_response=data,
        )

    def _requires_api_key(self) -> bool:
        return self.provider_api != "ollama-generate"

    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_messages(
        self,
        model: str,
        prompt: str,
        thinking: dict[str, Any] | None = None,
    ) -> 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,
        }
        resolved_thinking = self._resolve_anthropic_thinking(thinking)
        if resolved_thinking:
            request_body["thinking"] = resolved_thinking
        if self.provider == "minimax" and self.base_url.endswith("/anthropic"):
            path = "/v1/messages"
        else:
            path = "/messages"
        return await self._post(
            path=path,
            headers={
                "x-api-key": self.api_key,
                "anthropic-version": self.anthropic_version,
                "Content-Type": "application/json",
            },
            request_body=request_body,
        )

    def _resolve_anthropic_thinking(self, thinking: dict[str, Any] | None) -> dict[str, Any] | None:
        if thinking:
            return thinking

        # OpenClaw treats MiniMax's Anthropic-compatible path specially:
        # disable thinking by default unless the caller explicitly opts in.
        if self.provider == "minimax":
            return {"type": "disabled"}

        return None

    async def _request_anthropic_compatible(
        self,
        model: str,
        prompt: str,
        thinking: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        return await self._request_anthropic_messages(model, prompt, thinking)

    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_openai_blocks(self, payload: dict[str, Any]) -> list[AIContentBlock]:
        choices = payload.get("choices") or []
        if not choices:
            return []

        message = choices[0].get("message") or {}
        content = message.get("content")
        if isinstance(content, str):
            return [AIContentBlock(type="text", text=content)]
        if not isinstance(content, list):
            return []

        blocks: list[AIContentBlock] = []
        for item in content:
            if not isinstance(item, dict):
                continue
            blocks.append(
                AIContentBlock(
                    type=str(item.get("type", "text")),
                    text=item.get("text") if isinstance(item.get("text"), str) else None,
                    metadata={k: v for k, v in item.items() if k not in {"type", "text"}},
                )
            )
        return blocks
    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_anthropic_blocks(self, payload: dict[str, Any]) -> list[AIContentBlock]:
        content = payload.get("content")
        if not isinstance(content, list):
            return []

        blocks: list[AIContentBlock] = []
        for item in content:
            if not isinstance(item, dict):
                continue
            blocks.append(
                AIContentBlock(
                    type=str(item.get("type", "unknown")),
                    text=item.get("text") if isinstance(item.get("text"), str) else None,
                    thinking=item.get("thinking") if isinstance(item.get("thinking"), str) else None,
                    signature=item.get("signature") if isinstance(item.get("signature"), str) else None,
                    metadata={
                        k: v
                        for k, v in item.items()
                        if k not in {"type", "text", "thinking", "signature"}
                    },
                )
            )
        return blocks


    def _extract_ollama_content(self, payload: dict[str, Any]) -> str:
        response = payload.get("response")
        if isinstance(response, str):
            return response
        return ""

    def _extract_ollama_blocks(self, payload: dict[str, Any]) -> list[AIContentBlock]:
        response = payload.get("response")
        if isinstance(response, str) and response:
            return [AIContentBlock(type="text", text=response)]
        return []