topoteretes--cognee/cognee/infrastructure/session/feedback_detection.py

"""Session turn analysis from user messages via LLM."""

from cognee.infrastructure.llm.LLMGateway import LLMGateway
from cognee.infrastructure.llm.prompts import read_query_prompt
from cognee.infrastructure.session.feedback_models import SessionTurnAnalysis
from cognee.shared.logging_utils import get_logger

logger = get_logger("feedback_detection")


def _render_served_context(served_context) -> str:
    """Render served session-context entries as an ``id: content`` block.

    Accepts a pre-rendered string (returned as-is) or a list of dicts/objects exposing
    ``id``/``entry_id`` and ``content`` fields. Never raises: malformed items are skipped.
    """
    if served_context is None:
        return ""
    if isinstance(served_context, str):
        return served_context.strip()

    lines = []
    try:
        for item in served_context:
            if isinstance(item, dict):
                entry_id = item.get("id") or item.get("entry_id")
                content = item.get("content")
            else:
                entry_id = getattr(item, "id", None) or getattr(item, "entry_id", None)
                content = getattr(item, "content", None)
            if entry_id is None or content is None:
                continue
            lines.append(f"{str(entry_id).strip()}: {str(content).strip()}")
    except Exception:
        return ""
    return "\n".join(lines)


def _append_optional_section(text_input: str, title: str, content: str | None) -> str:
    if not content or not str(content).strip():
        return text_input
    return text_input + f"\n\n{title}:\n" + str(content).strip()


async def analyze_turn_for_session_context(
    user_message: str,
    *,
    previous_question: str | None = None,
    previous_answer: str | None = None,
    served_context: list | str | None = None,
) -> SessionTurnAnalysis:
    """
    Analyze a user message for answer routing and session-context updates.

    When ``served_context`` is provided (a pre-rendered string or a list of session-context
    entries served to the previous answer), it is appended to the LLM input so the single
    turn-analysis call can additionally produce ``served_context_ratings`` and
    ``candidate_context_updates``. This adds no extra LLM call.

    Returns a SessionTurnAnalysis. On LLM failure or timeout, returns an empty analysis so
    the main completion flow is never blocked.
    """
    if not (user_message and str(user_message).strip()):
        return SessionTurnAnalysis()

    try:
        system_prompt = read_query_prompt("feedback_detection_system.txt")
        if not system_prompt:
            logger.warning("Feedback detection: system prompt not found, skipping")
            return SessionTurnAnalysis()

        text_input = "CURRENT USER MESSAGE:\n" + user_message.strip()
        text_input = _append_optional_section(
            text_input,
            "PREVIOUS QUESTION",
            previous_question,
        )
        text_input = _append_optional_section(
            text_input,
            "PREVIOUS ANSWER",
            previous_answer,
        )
        rendered_context = _render_served_context(served_context)
        if rendered_context:
            text_input = (
                text_input
                + "\n\nSESSION CONTEXT ENTRIES SERVED TO THE PREVIOUS ANSWER (id: content):\n"
                + rendered_context
            )

        result = await LLMGateway.acreate_structured_output(
            text_input=text_input,
            system_prompt=system_prompt,
            response_model=SessionTurnAnalysis,
        )
        return result
    except Exception as e:
        logger.warning(
            "Session turn analysis failed, proceeding with empty analysis: %s",
            e,
            exc_info=False,
        )
        return SessionTurnAnalysis()


async def detect_feedback(
    user_message: str, served_context: list | str | None = None
) -> SessionTurnAnalysis:
    """Compatibility wrapper for older call sites."""
    return await analyze_turn_for_session_context(user_message, served_context=served_context)