api/src/agents/factory.py - master - ncit/multiagentsystem

api/src/agents/factory.py
91 строка3 KB
Nikita
feat: update API and add supabase initialization
09 июн 2026, 12:07
09 июн 2026, 12:07f18df6f
"""Ephemeral agent factory — creates agents on-demand per user.

Agents are stateless objects created per request.
Memory, knowledge, and session state persist in Postgres via Agno DB.
"""
from __future__ import annotations

from typing import Any

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.memory import MemoryManager
from agno.models.openrouter import OpenRouter

from src.agents.registry import AGENT_CONFIGS, AgentConfig
from src.agents.tools.registry import load_tools
from src.config import settings

# ─── Shared DB connection (not per-agent) ─────────────────────────────────────
_agent_db = PostgresDb(
    db_url=settings.database_url,
    session_table="agent_sessions",
    memory_table="agent_memory",
)


def create_agent(
    agent_type: str,
    user_id: str,
    *,
    session_id: str | None = None,
    model_id: str | None = None,
    instructions_override: str | None = None,
    tool_ids: list[str] | None = None,
    extra: dict[str, Any] | None = None,
) -> Agent:
    """Create an ephemeral Agno agent for a specific user.

    Args:
        agent_type: Key from AGENT_CONFIGS registry (e.g. 'coder', 'image-gen').
        user_id:    The authenticated user's ID (Supabase UUID).
        session_id: Optional session ID — auto-generated if None.
        model_id:   Override the default model (OpenRouter model slug).
        instructions_override: Replace the default system prompt.
        tool_ids:   Override the default tool set.
        extra:      Additional kwargs passed to Agent().

    Returns:
        A configured Agno Agent instance ready to run.
    """
    config: AgentConfig = AGENT_CONFIGS[agent_type]

    model = OpenRouter(
        id=model_id or config.model_id,
        api_key=settings.openrouter_api_key,
    )

    tools = load_tools(tool_ids or config.tool_ids)

    memory_manager = MemoryManager(
        model=model,
        db=_agent_db,
    )

    agent = Agent(
        name=f"{user_id}-{agent_type}",
        model=model,
        instructions=instructions_override or config.instructions,
        tools=tools,
        db=_agent_db,
        memory_manager=memory_manager,
        enable_agentic_memory=True,
        session_id=session_id or f"{user_id}-{agent_type}",
        markdown=True,
        **(extra or {}),
    )

    return agent


def list_agent_types() -> list[dict[str, str]]:
    """Return all available agent types from the registry."""
    return [
        {
            "type": key,
            "name": cfg.name,
            "description": cfg.description,
            "model_id": cfg.model_id,
        }
        for key, cfg in AGENT_CONFIGS.items()
    ]