python/backend/core/routing/intent_router.py - master - ksilisk/spbtechrun_hack

python/backend/core/routing/intent_router.py

40 строк2 KB

Shaliko Salimov

user profile support

12 дек 2025, 21:13

12 дек 2025, 21:1316730d4

from __future__ import annotations

from core.dialog.context_builder import ContextBuilder
from core.generators.answer_generator import AnswerGenerator
from core.retrieval.rag_chain import RagService
from schemas.chat import ChatRequest, Message
from schemas.intents import IntentType
from schemas.profile import UserProfile


class IntentRouter:
    def __init__(
            self,
            *,
            rag_service: RagService,
            answer_generator: AnswerGenerator,
            context_builder: ContextBuilder,
    ) -> None:
        self._rag_service = rag_service
        self._context_builder = context_builder
        self._answer_generator = answer_generator

    async def route(self, intent: IntentType, request: ChatRequest, history: list[Message], user_profile: UserProfile) -> str:
        if intent == IntentType.DYNAMIC_INFORMATION:
            system_prompt, trimmed_history = await self._context_builder.build_for_mcp(history=history, user_profile=user_profile)
            return await self._answer_generator.generate_using_mcp(
                trimmed_history,
                system_prompt,
                request.message,
            )
        elif intent == IntentType.STATIC_INFORMATION:
            rag_facts = await self._rag_service.ask(request.message)
            system_prompt, trimmed_history = await self._context_builder.build_for_rag(rag_facts, history, user_profile=user_profile)
            return await self._answer_generator.generate(trimmed_history, request.message, system_prompt)
        elif intent == IntentType.SMALL_TALK:
            system_prompt, trimmed_history = await self._context_builder.build_for_small_talk(history)
            return await self._answer_generator.generate(trimmed_history, request.message, system_prompt)
        else:
            system_prompt, trimmed_history = await self._context_builder.build_for_unknown_talk(history)
            return await self._answer_generator.generate(trimmed_history, request.message, system_prompt)