Merge branch 'llm-487' into wip

Author: nuwangeek

docs/CONTEXT_WORKFLOW_GREETING_DETECTION.md

@@ -60,6 +60,9 @@
 from src.utils.language_detector import detect_language, get_language_name
 from src.utils.prompt_config_loader import PromptConfigurationLoader
 from src.utils.query_validator import validate_query_basic
+from src.utils.sse_utils import extract_content_from_sse
+from src.utils.conversation_history_store import should_save_history, save_history_round
+from src.utils.conversation_history_helpers import get_conversation_history
 from src.guardrails import NeMoRailsAdapter, GuardrailCheckResult
 from src.contextual_retrieval import ContextualRetriever
 from src.contextual_retrieval.bm25_search import SmartBM25Search
@@ -68,13 +71,27 @@
     ContextualRetrievalFailureError,
 )
 from src.llm_orchestrator_config.feature_flags import FeatureFlags
-from src.tool_classifier import ToolClassifier
+from src.tool_classifier import ToolClassifier, WorkflowType
 from src.tool_classifier.constants import SERVICE_STEP_PREFIXES
 from src.tool_classifier.workflows.service_workflow import ServiceWorkflowExecutor
 
 # Initialize Loki logger for orchestration service
 logger = LokiLogger(service_name="llm-orchestration-service")
 
+# Set of content strings that must NOT be persisted in conversation history.
+# Covers all multilingual error / OOS / guardrail-violation messages so that
+# failed or blocked exchanges are never written to Redis.
+_HISTORY_EXCLUDED_MESSAGES: frozenset[str] = frozenset(
+    {
+        *OUT_OF_SCOPE_MESSAGES.values(),
+        *TECHNICAL_ISSUE_MESSAGES.values(),
+        *INPUT_GUARDRAIL_VIOLATION_MESSAGES.values(),
+        *OUTPUT_GUARDRAIL_VIOLATION_MESSAGES.values(),
+        *QUERY_VALIDATION_FAILED_MESSAGES.values(),
+        STREAM_TOKEN_LIMIT_MESSAGE,
+    }
+)
+
 
 class LangfuseConfig:
     """Configuration for Langfuse integration."""
@@ -152,6 +169,11 @@ def __init__(self) -> None:
         # Workflow executors access it via self.orchestration_service.session_store.
         self.session_store: Any = None
 
+        # Redis-backed conversation history store.
+        # Set to None here; the FastAPI lifespan injects the live store after
+        # Redis initialises (app.state.orchestration_service.conversation_history_store = ...).
+        self.conversation_history_store: Any = None
+
         # Shared BM25 search index pre-warmed at startup.
         # Populated by _prewarm_shared_bm25() which is called from the FastAPI
         # lifespan so it runs inside the async event loop.  Until then it is None
@@ -503,6 +525,18 @@ async def process_orchestration_request(
                     },
                 )
                 langfuse.flush()
+
+            # Persist successful exchange to conversation history (non-streaming)
+            if should_save_history(
+                self.conversation_history_store, response, _HISTORY_EXCLUDED_MESSAGES
+            ):
+                await save_history_round(
+                    self.conversation_history_store,
+                    request.chatId,
+                    request.message,
+                    response.content,
+                )
+
             return response
 
         except Exception as e:
@@ -731,14 +765,51 @@ async def stream_orchestration_response(
                         )
                         time_metric["classifier.route"] = time.time() - start_time
 
+                        # Accumulate content for history only on non-RAG workflows;
+                        # RAG routes through _stream_rag_pipeline which has its own hook.
+                        _save_classifier_history = (
+                            self.conversation_history_store is not None
+                            and classification.workflow != WorkflowType.RAG
+                        )
+                        _classifier_accumulated: list[str] = []
+                        # Tracks whether an excluded marker (OOS / guardrail violation /
+                        # error) was observed at any point during the stream.  When True
+                        # the entire accumulated buffer is discarded so no partial content
+                        # from before the blocked marker is ever written to Redis.
+                        _history_blocked = False
+
                         async for sse_chunk in stream_result:
                             yield sse_chunk
+                            if _save_classifier_history and not _history_blocked:
+                                extracted = extract_content_from_sse(sse_chunk)
+                                if extracted is not None and extracted != "END":
+                                    if extracted in _HISTORY_EXCLUDED_MESSAGES:
+                                        # Excluded marker observed — discard any partial
+                                        # content accumulated before this point and stop
+                                        # accumulating for the rest of the stream.
+                                        _classifier_accumulated.clear()
+                                        _history_blocked = True
+                                    else:
+                                        _classifier_accumulated.append(extracted)
 
                         # Successfully completed streaming through classifier
                         logger.info(
                             f"[{request.chatId}] [{stream_ctx.stream_id}] Tool classifier streaming completed"
                         )
 
+                        # Persist conversation history (classifier streaming, non-RAG workflows)
+                        if (
+                            _save_classifier_history
+                            and not _history_blocked
+                            and _classifier_accumulated
+                        ):
+                            await save_history_round(
+                                self.conversation_history_store,
+                                request.chatId,
+                                request.message,
+                                "".join(_classifier_accumulated),
+                            )
+
                         # Log costs and timings
                         self.log_costs(costs_metric)
                         log_step_timings(time_metric, request.chatId)
@@ -868,10 +939,16 @@ async def _stream_rag_pipeline(
         )
 
         start_time = time.time()
+        conversation_history, conversation_summary = await get_conversation_history(
+            chat_id=request.chatId,
+            store=self.conversation_history_store,
+            fallback=request.conversationHistory,
+        )
         refined_output, refiner_usage = self._refine_user_prompt(
             llm_manager=components["llm_manager"],
             original_message=request.message,
-            conversation_history=request.conversationHistory,
+            conversation_history=conversation_history,
+            conversation_summary=conversation_summary,
         )
         time_metric["prompt_refiner"] = time.time() - start_time
         costs_metric["prompt_refiner"] = refiner_usage
@@ -1173,6 +1250,17 @@ async def bot_response_generator() -> AsyncIterator[str]:
                     f"Storage failed for chat_id: {request.chatId}, environment: {request.environment} - {str(storage_error)}"
                 )
 
+            # Persist conversation history (RAG streaming)
+            if self.conversation_history_store is not None:
+                _rag_bot_message = "".join(accumulated_response)
+                if _rag_bot_message not in _HISTORY_EXCLUDED_MESSAGES:
+                    await save_history_round(
+                        self.conversation_history_store,
+                        request.chatId,
+                        request.message,
+                        _rag_bot_message,
+                    )
+
             # Mark stream as completed successfully
             stream_ctx.mark_completed()
 
@@ -1422,10 +1510,16 @@ async def _execute_orchestration_pipeline(
 
         # Step 1: Refine user prompt
         start_time = time.time()
+        conversation_history, conversation_summary = await get_conversation_history(
+            chat_id=request.chatId,
+            store=self.conversation_history_store,
+            fallback=request.conversationHistory,
+        )
         refined_output, refiner_usage = self._refine_user_prompt(
             llm_manager=components["llm_manager"],
             original_message=request.message,
-            conversation_history=request.conversationHistory,
+            conversation_history=conversation_history,
+            conversation_summary=conversation_summary,
         )
         timing_key = f"{prefix}.prompt_refiner" if prefix else "prompt_refiner"
         time_metric[timing_key] = time.time() - start_time
@@ -2304,6 +2398,7 @@ def _refine_user_prompt(
         llm_manager: LLMManager,
         original_message: str,
         conversation_history: List[ConversationItem],
+        conversation_summary: Optional[str] = None,
     ) -> tuple[PromptRefinerOutput, Dict[str, Any]]:
         """
         Refine user prompt using loaded LLM configuration and return usage info.
@@ -2312,6 +2407,10 @@ def _refine_user_prompt(
             llm_manager: The LLM manager instance to use
             original_message: The original user message to refine
             conversation_history: Previous conversation context
+            conversation_summary: Optional summary of earlier conversation rounds
+                that were evicted from Redis.  When provided it is prepended to the
+                DSPy history as a ``system`` turn so the refiner can use it for
+                context without re-summarising via an LLM call.
 
         Returns:
             Tuple of (PromptRefinerOutput, usage_dict): The refined prompt output and usage info
@@ -2324,8 +2423,16 @@ def _refine_user_prompt(
         logger.info("Starting prompt refinement process")
 
         try:
-            # Convert conversation history to DSPy format
+            # Convert conversation history to DSPy format, optionally prepending
+            # a pre-computed summary of earlier (evicted) conversation rounds.
             history: List[Dict[str, str]] = []
+            if conversation_summary:
+                history.append(
+                    {
+                        "role": "system",
+                        "content": f"Summary of earlier conversation: {conversation_summary}",
+                    }
+                )
             for item in conversation_history:
                 role = "assistant" if item.authorRole == "bot" else item.authorRole
                 history.append({"role": role, "content": item.message})

src/llm_orchestration_service.py

@@ -1,5 +1,6 @@
 """LLM Orchestration Service API - FastAPI application."""
 
+import asyncio
 import os
 import logging
 from contextlib import asynccontextmanager
@@ -12,12 +13,15 @@
 import uvicorn
 
 from llm_orchestration_service import LLMOrchestrationService
+from llm_orchestrator_config.llm_manager import LLMManager
 from src.utils.redis_client import (
     init_redis_client,
     close_redis_client,
     check_redis_health,
 )
 from src.utils.api_tool_session_store import APIToolSessionStore
+from src.utils.conversation_history_store import ConversationHistoryStore
+from src.utils.conversation_summary_generator import create_incremental_summarizer
 from src.llm_orchestrator_config.llm_ochestrator_constants import (
     STREAMING_ALLOWED_ENVS,
     STREAM_TIMEOUT_MESSAGE,
@@ -98,23 +102,50 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     try:
         await init_redis_client()
         app.state.session_store = APIToolSessionStore()
+
+        # Wire an incremental summarizer if the LLM manager singleton is available.
+        summarizer = None
+        try:
+            summarizer = create_incremental_summarizer(LLMManager())
+            logger.info("Incremental conversation summarizer initialized")
+        except Exception as e:
+            logger.warning(
+                f"Could not create incremental summarizer, continuing without it: {e}"
+            )
+
+        app.state.conversation_history_store = ConversationHistoryStore(
+            summarizer=summarizer
+        )
         logger.info("Redis session store initialized successfully")
     except Exception as e:
         logger.warning(f"Redis session store unavailable, continuing without it: {e}")
         app.state.session_store = None
+        app.state.conversation_history_store = None
 
-    # Expose session_store on the orchestration service so workflow executors
-    # (e.g. APIToolWorkflowExecutor) can reach it via self.orchestration_service.
+    # Expose session_store and conversation_history_store on the orchestration
+    # service so downstream components can reach them via self.orchestration_service.
     if (
         hasattr(app.state, "orchestration_service")
         and app.state.orchestration_service is not None
     ):
         app.state.orchestration_service.session_store = app.state.session_store
+        app.state.orchestration_service.conversation_history_store = (
+            app.state.conversation_history_store
+        )
 
     yield
 
     # Shutdown
     logger.info("Shutting down LLM Orchestration Service API")
+
+    # Await any in-flight incremental summary tasks to avoid lost work.
+    store = getattr(app.state, "conversation_history_store", None)
+    if store is not None and store._pending_tasks:
+        logger.info(
+            f"Waiting for {len(store._pending_tasks)} pending summary task(s) to complete..."
+        )
+        await asyncio.gather(*store._pending_tasks, return_exceptions=True)
+
     if (
         hasattr(app.state, "orchestration_service")
         and app.state.orchestration_service is not None

src/llm_orchestration_service_api.py

@@ -0,0 +1,45 @@
+"""Pydantic models for conversation history state."""
+
+import time
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class ConversationRound(BaseModel):
+    """A single user+bot exchange in a conversation.
+
+    Stored as part of ``ConversationHistoryState`` in Redis, keyed by chat_id.
+    """
+
+    user_message: str = Field(..., description="The user's message text for this round")
+    bot_message: str = Field(..., description="The bot's response text for this round")
+    timestamp: float = Field(
+        default_factory=time.time,
+        description="Unix timestamp of when the round was recorded",
+    )
+
+
+class ConversationHistoryState(BaseModel):
+    """Persisted conversation history for a chat session.
+
+    Keyed by chat_id in Redis with a sliding 30-minute TTL.
+    Retains up to the most recent 10 rounds; older rounds are trimmed.
+    An optional summary field holds a condensed representation of rounds
+    that have been evicted. When a summarizer is injected into the store,
+    it is automatically generated and persisted as rounds are evicted.
+    If no summarizer is provided, the summary must be managed by the caller.
+    """
+
+    chat_id: str = Field(..., description="Unique conversation identifier")
+    rounds: list[ConversationRound] = Field(
+        default_factory=list,
+        description="Ordered list of conversation rounds (newest last), capped at 10",
+    )
+    summary: Optional[str] = Field(
+        default=None,
+        description=(
+            "Optional condensed summary of earlier conversation turns that have been "
+            "evicted from the rounds list. Generated and stored by the caller."
+        ),
+    )

src/models/conversation_history_models.py

@@ -127,6 +127,9 @@ def __init__(
         self.context_workflow = ContextWorkflowExecutor(
             llm_manager=llm_manager,
             orchestration_service=orchestration_service,
+            conversation_history_store=getattr(
+                orchestration_service, "conversation_history_store", None
+            ),
         )
         self.rag_workflow = RAGWorkflowExecutor(
             orchestration_service=orchestration_service,