76 lines
3.9 KiB
Python
76 lines
3.9 KiB
Python
# rag/agents/coordinator.py
|
||
"""
|
||
Координатор агентов: управляет специализированными агентами и маршрутизирует запросы.
|
||
"""
|
||
|
||
import logging
|
||
from typing import Dict, List, Optional, Any
|
||
|
||
from rag.agents.roles import BaseAgent
|
||
from rag.services.giga_client import GigaClient
|
||
from rag.config_models import AppConfig
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AgentCoordinator:
|
||
"""
|
||
Координатор, который выбирает подходящего агента для запроса и управляет
|
||
диалогом между агентами (при необходимости).
|
||
"""
|
||
|
||
def __init__(self, config: AppConfig, giga: GigaClient):
|
||
self.config = config
|
||
self.giga = giga
|
||
self.agents: Dict[str, BaseAgent] = {}
|
||
|
||
def register_agent(self, name: str, agent: BaseAgent) -> None:
|
||
"""Регистрирует агента с указанным именем."""
|
||
self.agents[name] = agent
|
||
logger.info(f"Агент '{name}' зарегистрирован")
|
||
|
||
async def process(
|
||
self,
|
||
query: str,
|
||
context: str,
|
||
history: List[Dict[str, str]],
|
||
tools: Optional[List[Dict[str, Any]]] = None,
|
||
) -> str:
|
||
"""
|
||
Основной метод: выбирает агента и передаёт ему запрос.
|
||
Если агент возвращает действие, выполняет его (или передаёт другому агенту).
|
||
"""
|
||
# Простая маршрутизация: определяем ключевые слова
|
||
query_lower = query.lower()
|
||
# Можно использовать классификатор намерений или просто по ключевым словам
|
||
if any(kw in query_lower for kw in ["методолог", "стандарт", "гост", "методика", "регламент"]):
|
||
agent_name = "methodologist"
|
||
elif any(kw in query_lower for kw in ["коробочное", "типовое", "решение", "шаблон", "создай документ"]):
|
||
agent_name = "box_solution"
|
||
elif any(kw in query_lower for kw in ["командировка", "бронирование", "билет", "отель", "календарь"]):
|
||
agent_name = "personal_assistant"
|
||
else:
|
||
# Если не определили, можно использовать первый зарегистрированный или общий
|
||
agent_name = "methodologist" # по умолчанию
|
||
|
||
agent = self.agents.get(agent_name)
|
||
if not agent:
|
||
logger.warning(f"Агент '{agent_name}' не найден, используем первого доступного")
|
||
if self.agents:
|
||
agent = next(iter(self.agents.values()))
|
||
else:
|
||
return "Нет доступных агентов для обработки запроса."
|
||
|
||
# Обрабатываем запрос агентом
|
||
result = await agent.process(query, context, history, tools)
|
||
|
||
if result.get("type") == "answer":
|
||
return result.get("content", "")
|
||
elif result.get("type") == "action":
|
||
# Если агент вернул действие (например, вызов инструмента), можно выполнить его
|
||
# и передать результат другому агенту или тому же самому.
|
||
# Для простоты мы просто вернём сообщение о необходимости действия.
|
||
action = result.get("content", {})
|
||
return f"Агент запросил действие: {action}"
|
||
else:
|
||
return "Неизвестный тип ответа от агента." |