conversation history structure

app/main.py

@@ -66,22 +66,40 @@ compiled_graph = workflow.compile()
 #----------------------------------------
 
 async def chatui_adapter(data):
-    """Text-only adapter for ChatUI"""
+    """Text-only adapter for ChatUI with structured message support"""
     try:
-        text = data.text if hasattr(data, 'text') else data.get('text', '')
-
-        logger.info(f"Processing query: {text}")
+        # Extract query - prefer structured messages over legacy text field
+        if hasattr(data, 'messages') and data.messages:
+            messages = data.messages
+            # Extract latest user query
+            user_messages = [msg for msg in messages if msg.role == 'user']
+            query = user_messages[-1].content if user_messages else ""
+            
+            # Log conversation context
+            logger.info(f"Processing query: {query}")
+            logger.info(f"Total messages in conversation: {len(messages)}")
+            logger.info(f"User messages: {len(user_messages)}, Assistant messages: {len([m for m in messages if m.role == 'assistant'])}")
+            
+            # Optional: Build conversation context for generation (last N turns)
+            conversation_context = build_conversation_context(messages, max_turns=3)
+            logger.info(f"Conversation context: {len(conversation_context)} characters")
+        else:
+            # Fallback to legacy text field
+            query = data.text if hasattr(data, 'text') else data.get('text', '')
+            conversation_context = None
+            logger.info(f"Processing query (legacy): {query}")
         
         full_response = ""
         sources_collected = None
         
         async for result in process_query_streaming(
-            query=text,
+            query=query,
             file_upload=None,
             reports_filter="",
             sources_filter="",
             subtype_filter="",
-            year_filter=""
+            year_filter="",
+            conversation_context=conversation_context  # Pass to processing function
         ):
             if isinstance(result, dict):
                 result_type = result.get("type", "data")
@@ -111,9 +129,23 @@ async def chatui_adapter(data):
 
 
 async def chatui_file_adapter(data):
-    """File upload adapter for ChatUI (LangServe endpoint)"""
+    """File upload adapter for ChatUI with structured message support"""
     try:
-        text = data.text if hasattr(data, 'text') else data.get('text', '')
+        # Extract query - prefer structured messages
+        if hasattr(data, 'messages') and data.messages:
+            messages = data.messages
+            user_messages = [msg for msg in messages if msg.role == 'user']
+            query = user_messages[-1].content if user_messages else ""
+            
+            logger.info(f"Processing query: {query}")
+            logger.info(f"Total messages: {len(messages)}")
+            
+            conversation_context = build_conversation_context(messages, max_turns=3)
+        else:
+            query = data.text if hasattr(data, 'text') else data.get('text', '')
+            conversation_context = None
+            logger.info(f"Processing query (legacy): {query}")
+        
         files = getattr(data, 'files', None) if hasattr(data, 'files') else data.get('files', None)
         
         file_content = None
@@ -135,14 +167,15 @@ async def chatui_file_adapter(data):
         sources_collected = None
         
         async for result in process_query_streaming(
-            query=text,
+            query=query,
             file_content=file_content,
             filename=filename,
             reports_filter="",
             sources_filter="",
             subtype_filter="",
             year_filter="",
-            output_format="structured"
+            output_format="structured",
+            conversation_context=conversation_context
         ):
             if isinstance(result, dict):
                 result_type = result.get("type", "data")
@@ -153,7 +186,6 @@ async def chatui_file_adapter(data):
                 elif result_type == "sources":
                     sources_collected = content
                 elif result_type == "end":
-                    # Send sources at the end, like the text-only adapter
                     if sources_collected:
                         sources_text = "\n\n**Sources:**\n"
                         for i, source in enumerate(sources_collected, 1):
@@ -176,6 +208,45 @@ async def chatui_file_adapter(data):
         yield f"Error: {str(e)}"
 
 
+def build_conversation_context(messages: List, max_turns: int = 3, max_chars: int = 2000) -> str:
+    """
+    Build conversation context from structured messages.
+    Keeps the most recent turns within character budget.
+    
+    Args:
+        messages: List of Message objects
+        max_turns: Maximum number of conversation turns (user+assistant pairs) to include
+        max_chars: Maximum total characters in context
+    """
+    context_parts = []
+    char_count = 0
+    turn_count = 0
+    
+    # Process messages in reverse to keep most recent
+    for msg in reversed(messages):
+        role_label = msg.role.upper()
+        content = msg.content
+        
+        # Estimate if adding this message would exceed limits
+        msg_text = f"{role_label}: {content}"
+        msg_chars = len(msg_text)
+        
+        if char_count + msg_chars > max_chars:
+            break
+        
+        if msg.role in ['user', 'assistant']:
+            if msg.role == 'user':
+                turn_count += 1
+            if turn_count > max_turns:
+                break
+        
+        context_parts.insert(0, msg_text)
+        char_count += msg_chars
+    
+    context = "\n\n".join(context_parts)
+    logger.info(f"Built conversation context: {turn_count} turns, {char_count} chars")
+    return context
+
 #----------------------------------------
 # FASTAPI SETUP - for future use
 #----------------------------------------

app/models.py

@@ -23,9 +23,13 @@ class GraphState(TypedDict):
 
 class ChatUIInput(BaseModel):
     """Input model for text-only ChatUI requests"""
-    text: str
+    text: str  # Legacy: full concatenated prompt (for backward compatibility)
+    messages: Optional[List[Message]] = None  # Structured conversation history
+    preprompt: Optional[str] = None
 
 class ChatUIFileInput(BaseModel):
     """Input model for ChatUI requests with file attachments"""
     text: str
     files: Optional[List[Dict[str, Any]]] = None
+    messages: Optional[List[Message]] = None  # Structured conversation history
+    preprompt: Optional[str] = None

app/nodes.py

@@ -403,11 +403,17 @@ async def process_query_streaming(
     reports_filter: str = "", 
     sources_filter: str = "", 
     subtype_filter: str = "", 
-    year_filter: str = "", 
-    output_format: str = "structured"
+    year_filter: str = "",
+    output_format: str = "structured",
+    conversation_context: Optional[str] = None  # NEW: conversation context
 ):
     """
-    Unified streaming function supporting both file objects and raw content.
+    Unified streaming function with conversation context support.
+    
+    Args:
+        query: Latest user query
+        conversation_context: Optional conversation history for generation context
+        ... (other args remain the same)
     """
     # Handle file_upload if provided
     if file_upload is not None:
@@ -427,10 +433,15 @@ async def process_query_streaming(
     start_time = datetime.now()
     session_id = f"stream_{start_time.strftime('%Y%m%d_%H%M%S')}"
     
+    # Log retrieval strategy
+    logger.info(f"Retrieval query: {query[:100]}...")
+    if conversation_context:
+        logger.info(f"Generation will use conversation context ({len(conversation_context)} chars)")
+    
     try:
         # Build initial state
         initial_state = {
-            "query": query,
+            "query": query,  # Use ONLY latest query for retrieval
             "context": "",
             "ingestor_context": "",
             "result": "",
@@ -443,23 +454,23 @@ async def process_query_streaming(
             "filename": filename,
             "file_type": "unknown",
             "workflow_type": "standard",
+            "conversation_context": conversation_context,  # Store for generation
             "metadata": {
                 "session_id": session_id,
                 "start_time": start_time.isoformat(),
-                "has_file_attachment": file_content is not None
+                "has_file_attachment": file_content is not None,
+                "has_conversation_context": conversation_context is not None
             }
         }
         
         # Execute workflow nodes
         if file_content and filename:
-            # File present: detect type and process
             state = merge_state(initial_state, detect_file_type_node(initial_state))
             state = merge_state(state, ingest_node(state))
             
             workflow_type = route_workflow(state)
             
             if workflow_type == "direct_output":
-                # NEW file with direct output enabled: show results and return
                 final_state = direct_output_node(state)
                 if output_format == "structured":
                     yield {"type": "data", "content": final_state["result"]}
@@ -468,13 +479,14 @@ async def process_query_streaming(
                     yield final_state["result"]
                 return
             else:
-                # Standard workflow or cached direct output: continue to RAG
+                # Retrieve using ONLY the latest query
                 state = merge_state(state, retrieve_node(state))
         else:
-            # No file: standard RAG pipeline
+            # No file: retrieve using latest query only
             state = merge_state(initial_state, retrieve_node(initial_state))
         
         # Generate response with streaming
+        # The generator can optionally use conversation_context for better responses
         sources_collected = None
         accumulated_response = "" if output_format == "gradio" else None