vent_ai_platform/app/agents/runtime.py

from __future__ import annotations

import asyncio
import importlib.util
from dataclasses import dataclass
from functools import lru_cache
from typing import Any

from app.agents.prompts import VENT_SYSTEM_PROMPT
from app.core.config import settings
from app.core.logging import setup_logging


@dataclass
class AgentInvocationResult:
    status: str
    output: str
    model: str
    error: str = ""


class DeepAgentRuntime:
    def __init__(self) -> None:
        self._agent: Any | None = None
        self._lock = asyncio.Lock()

    @property
    def dependencies_available(self) -> bool:
        return all(
            importlib.util.find_spec(name) is not None
            for name in ("deepagents", "langchain_openai", "langgraph")
        )

    @property
    def enabled(self) -> bool:
        return self.dependencies_available and settings.model_configured

    def describe(self) -> dict[str, Any]:
        return {
            "dependencies_available": self.dependencies_available,
            "model_configured": settings.model_configured,
            "enabled": self.enabled,
            "model": settings.deepagent_model,
            "framework": "deepagents",
        }

    async def _build_agent(self) -> Any:
        if self._agent is not None:
            return self._agent
        async with self._lock:
            if self._agent is not None:
                return self._agent
            from deepagents import create_deep_agent
            from langchain_openai import ChatOpenAI

            model = ChatOpenAI(
                model=settings.deepagent_model,
                api_key=settings.openai_api_key,
                base_url=settings.openai_base_url or None,
                timeout=settings.openai_timeout,
                max_retries=2,
            )
            self._agent = create_deep_agent(
                model=model,
                tools=[],
                system_prompt=VENT_SYSTEM_PROMPT,
                name="vent-platform-agent",
            )
            return self._agent

    async def invoke(
        self,
        *,
        message: str,
        thread_id: str,
        context: dict[str, Any] | None = None,
    ) -> AgentInvocationResult:
        if not self.enabled:
            return AgentInvocationResult(
                status="disabled",
                output="",
                model=settings.deepagent_model,
                error="DeepAgent 依赖或模型配置未就绪",
            )
        try:
            agent = await self._build_agent()
            payload = message
            if context:
                payload = f"{message}\n\n业务上下文：\n{context}"
            result = await asyncio.wait_for(
                agent.ainvoke(
                    {"messages": [{"role": "user", "content": payload}]},
                    config={"configurable": {"thread_id": thread_id}},
                ),
                timeout=settings.openai_timeout,
            )
            messages = result.get("messages", []) if isinstance(result, dict) else []
            content = messages[-1].content if messages else str(result)
            if not isinstance(content, str):
                content = str(content)
            return AgentInvocationResult(
                status="success",
                output=content,
                model=settings.deepagent_model,
            )
        except Exception as exc:  # pragma: no cover - 依赖外部模型
            setup_logging()
            return AgentInvocationResult(
                status="failed",
                output="",
                model=settings.deepagent_model,
                error=f"{type(exc).__name__}: {exc}",
            )


@lru_cache(maxsize=1)
def get_agent_runtime() -> DeepAgentRuntime:
    return DeepAgentRuntime()