dpb/backend/app/utils/ai_factory.py

from typing import Any

from agno.agent import Agent
from agno.models.openai.like import OpenAILike
from agno.team import Team

from app.config.setting import settings


class AgnoFactory:
    """Agno 工厂类 - 统一管理 Agent、Team 创建逻辑"""

    # 配置常量
    AGENT_DESCRIPTION = "你是一个有用的AI助手，可以帮助用户回答问题和提供帮助。"
    AGENT_INSTRUCTIONS = ["保持回答简洁明了", "如果不确定，请说明"]
    AGENT_EXPECTED_OUTPUT = "中文回答"
    AGENT_TEMPERATURE = 0.7
    NUM_HISTORY_RUNS = 3
    REQUEST_TIMEOUT = 60.0  # LLM 请求总超时（秒），流式响应需放长
    CONNECT_TIMEOUT = 10.0  # TCP 连接超时（秒）

    def create_agent(
        self,
        user_id: str,
        team_id: str,
        session_id: str,
        db: Any | None = None,
        model_config: dict[str, Any] | None = None,
    ) -> Team:
        """创建带 Agent 的 Team 实例。

        参数:
        - user_id (str): 用户标识。
        - team_id (str): 团队标识。
        - session_id (str): 会话 ID。
        - db (Any | None): Agno 持久化数据库实例，可选。
        - model_config (dict | None): 运行时模型配置，覆盖系统默认。
            支持字段：base_url, api_key, model_id, temperature。

        返回:
        - Team: 配置好的 Team。
        """
        # 优先使用运行时配置，否则 fallback 到系统 settings
        base_url = settings.OPENAI_BASE_URL
        api_key = settings.OPENAI_API_KEY
        model_id = settings.OPENAI_MODEL
        temperature = self.AGENT_TEMPERATURE

        if model_config:
            base_url = model_config.get("base_url") or base_url
            api_key = model_config.get("api_key") or api_key
            model_id = model_config.get("model_id") or model_id
            if isinstance(model_config.get("temperature"), (int, float)):
                temperature = float(model_config["temperature"])

        # 创建 Agent
        dpb_agent = Agent(
            id=user_id,
            name="dpb_agent",
            role="You are a helpful AI assistant",
            description=self.AGENT_DESCRIPTION,
            tools=[],
        )

        # 创建 Team
        dpb_team = Team(
            id=team_id,
            user_id=user_id,
            session_id=session_id,
            model=OpenAILike(
                id=model_id,
                api_key=api_key,
                base_url=base_url,
                temperature=temperature,
                timeout=self.REQUEST_TIMEOUT,
            ),
            members=[dpb_agent],
            instructions=self.AGENT_INSTRUCTIONS,
            expected_output=self.AGENT_EXPECTED_OUTPUT,
            add_datetime_to_context=True,
            add_history_to_context=True,
            markdown=True,
            num_history_runs=self.NUM_HISTORY_RUNS,
            input_schema=None,
            output_schema=None,
            parse_response=True,
            read_chat_history=True,
            db=db,
        )

        return dpb_team