def generate_case_signal_historical_context()

in src/dispatch/ai/service.py [0:0]

• 62 lines of code
• 17 McCabe index (conditional complexity)

def generate_case_signal_historical_context(case: Case, db_session: Session) -> str:
    """
    Generate historical context for a case stemming from a signal, including related cases and relevant data.

    Args:
        case (Case): The case object for which historical context is being generated.
        db_session (Session): The database session used for querying related data.

    Returns:
        str: A string containing the historical context for the case, or an error message if context generation fails.
    """
    # we fetch the first instance id and signal
    (first_instance_id, first_instance_signal) = signal_service.get_instances_in_case(
        db_session=db_session, case_id=case.id
    ).first()

    signal_instance = signal_service.get_signal_instance(
        db_session=db_session, signal_instance_id=first_instance_id
    )

    # Check if the signal instance is valid
    if not signal_instance:
        message = "Unable to generate historical context. Signal instance not found."
        log.warning(message)
        raise GenAIException(message)

    # Check if the signal is valid
    if not signal_instance.signal:
        message = "Unable to generate historical context. Signal not found."
        log.warning(message)
        raise GenAIException(message)

    # Check if GenAI is enabled for the signal
    if not signal_instance.signal.genai_enabled:
        message = (
            "Unable to generate historical context. GenAI feature not enabled for this detection."
        )
        log.warning(message)
        raise GenAIException(message)

    # we fetch related cases
    related_cases = []
    for resolution_reason in CaseResolutionReason:
        related_cases.extend(
            signal_service.get_cases_for_signal_by_resolution_reason(
                db_session=db_session,
                signal_id=first_instance_signal.id,
                resolution_reason=resolution_reason,
            )
            .from_self()  # NOTE: function deprecated in SQLAlchemy 1.4 and removed in 2.0
            .filter(Case.id != case.id)
        )

    # we prepare historical context
    historical_context = []
    for related_case in related_cases:
        historical_context.append("<case>")
        historical_context.append(f"<case_name>{related_case.name}</case_name>")
        historical_context.append(f"<case_resolution>{related_case.resolution}</case_resolution")
        historical_context.append(
            f"<case_resolution_reason>{related_case.resolution_reason}</case_resolution_reason>"
        )
        historical_context.append(
            f"<case_alert_data>{related_case.signal_instances[0].raw}</case_alert_data>"
        )
        conversation_plugin = plugin_service.get_active_instance(
            db_session=db_session, project_id=case.project.id, plugin_type="conversation"
        )
        if conversation_plugin:
            if related_case.conversation and related_case.conversation.channel_id:
                # we fetch conversation replies for the related case
                conversation_replies = conversation_plugin.instance.get_conversation_replies(
                    conversation_id=related_case.conversation.channel_id,
                    thread_ts=related_case.conversation.thread_id,
                )
                for reply in conversation_replies:
                    historical_context.append(
                        f"<case_conversation_reply>{reply}</case_conversation_reply>"
                    )
        else:
            log.warning(
                "Conversation replies not included in historical context. No conversation plugin enabled."
            )
        historical_context.append("</case>")

    return "\n".join(historical_context)