fix: improve nudges messages by getting more context from messages (#1617)

* improve nudges fetching

* style: ruff format (auto)

* style: ruff autofix (auto)

* fixed coderabbit suggestions

* style: ruff autofix (auto)

* fix ruff

* fix: apply CodeRabbit auto-fixes

Fixed 1 file(s) based on 1 unresolved review comment.

Co-authored-by: CodeRabbit <noreply@coderabbit.ai>

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: CodeRabbit <noreply@coderabbit.ai>

Author: 4 people

flows/openrag_nudges.json

@@ -283,13 +283,18 @@
             "display_name": "Prompt Template",
             "documentation": "https://docs.langflow.org/components-prompts",
             "edited": false,
+            "error": null,
             "field_order": [
               "template",
               "use_double_brackets",
               "tool_placeholder"
             ],
             "frozen": false,
+            "full_path": null,
             "icon": "prompts",
+            "is_composition": null,
+            "is_input": null,
+            "is_output": null,
             "legacy": false,
             "metadata": {
               "code_hash": "3d3fd7b8a36f",
@@ -305,6 +310,7 @@
               "module": "custom_components.prompt_template"
             },
             "minimized": false,
+            "name": "",
             "output_types": [],
             "outputs": [
               {
@@ -327,6 +333,8 @@
               }
             ],
             "pinned": false,
+            "priority": null,
+            "replacement": null,
             "template": {
               "_type": "Component",
               "code": {
@@ -410,7 +418,7 @@
                 "trace_as_input": true,
                 "track_in_telemetry": false,
                 "type": "prompt",
-                "value": "You are generating prompt nudges to help a user explore a corpus.\n\nTask:\n1) Skim the documents to infer common themes, entities, or tasks.\n2) Propose exactly three concise, distinct prompt nudges that encourage useful next queries.\n3) If the chat history is provided, use it to generate new questions that the user might have, based on the llm's response to his previous query. DO NOT repeat user questions.\n4) Make the nudges concise, close to 40 characters.\n5) The nudges are questions or commands that the user can make to the chatbot, which will respond looking at the corpus.\n4) Return strings only, separating the nudges by a newline. Don't include quotation marks.\n5) If any error occured, return blank. This will be used in production, so don't ask for more info or confirm a info like you're talking to me still. If, for some reason, you can't provide the nudges, your job failed and you just return blank.\nRules: Be brief. No duplicates. No explanations outside the strings of the nudges. English only.\n\nExamples:\n  Show me this quarter's top 10 deals\n  Summarize recent client interactions\n  Search OpenSearch for mentions of our competitors\n\nChat history:\n{prompt}\n\nDocuments:\n{docs}\n\n"
+                "value": "You are generating prompt nudges to help a user explore a corpus.\n\nTask:\n1) Skim the documents to infer common themes, entities, or tasks.\n2) Propose exactly three concise, distinct prompt nudges (close to 40 characters each) that encourage useful next queries.\n3) Ensure the nudges are questions or commands that the user can send to the chatbot.\n4) Return strings only, separated by a newline. Do not include quotation marks.\n5) If an error occurs, return a blank string.\n\nRules: Be brief. No duplicates. No explanations outside the strings of the nudges. English only.\n\nIf chat history is provided, follow these rules:\n1) Generate recommendations based solely on the chat history and its chunks, not on the documents.\n2) Generate new questions that the user might have based on the LLM's response to their previous query. DO NOT repeat user questions.\n3) Match the style of the previous questions (e.g., if the user asked a question, generate nudges as questions).\n\nExamples:\n  What are this quarter's top 10 deals?\n  Summarize recent client interactions\n  Search OpenSearch for mentions of our competitors\n\n--------------------------------------------------------\n\nChat history:\n{prompt}\n\n\n--------------------------------------------------------\n\nDocuments (ignore if chat history is not empty):\n{docs}"
               },
               "tool_placeholder": {
                 "_input_type": "MessageTextInput",
@@ -759,14 +767,14 @@
             },
             "tool_mode": false
           },
-          "showNode": false,
+          "showNode": true,
           "type": "ChatInput"
         },
         "dragging": false,
         "id": "ChatInput-7W1BE",
         "measured": {
-          "height": 52,
-          "width": 192
+          "height": 207,
+          "width": 320
         },
         "position": {
           "x": 908.3744348370491,

src/services/chat_service.py

@@ -320,20 +320,140 @@ async def langflow_nudges_chat(
             )
         prompt = ""
         if previous_response_id:
-            from agent import get_conversation_thread
-
-            conversation_history = get_conversation_thread(
-                conversation_user_id, previous_response_id
-            )
-            if conversation_history:
-                conversation_history = "\n".join(
-                    [
-                        f"{msg['role']}: {msg['content']}"
-                        for msg in conversation_history["messages"]
-                        if msg["role"] in ["user", "assistant"]
-                    ]
+            messages = []
+            # Try in-memory active conversation first
+            from agent import active_conversations
+
+            if (
+                conversation_user_id in active_conversations
+                and previous_response_id in active_conversations[conversation_user_id]
+            ):
+                messages = active_conversations[conversation_user_id][previous_response_id].get(
+                    "messages", []
                 )
-                prompt = f"{conversation_history}"
+
+            # Filter out system messages
+            user_ast_messages = [m for m in messages if m.get("role") in ["user", "assistant"]]
+
+            # If no history in memory, try fetching from Langflow persistent history
+            if not user_ast_messages:
+                from services.langflow_history_service import langflow_history_service
+
+                try:
+                    lf_messages = await langflow_history_service.get_session_messages(
+                        conversation_user_id, previous_response_id
+                    )
+                    if lf_messages:
+                        messages = lf_messages
+                        user_ast_messages = [
+                            m for m in messages if m.get("role") in ["user", "assistant"]
+                        ]
+                except Exception as e:
+                    logger.warning(f"Failed to fetch session messages for nudges: {e}")
+
+            if user_ast_messages:
+                # Return at max 8 messages (the last ones)
+                user_ast_messages = user_ast_messages[-8:]
+
+                formatted_messages = []
+
+                def trim_results(r, depth=0):
+                    MAX_DEPTH = 3
+                    MAX_LIST_LEN = 3
+                    MAX_DICT_KEYS = 5
+                    MAX_STR_LEN = 1000
+
+                    if isinstance(r, str):
+                        return r[:MAX_STR_LEN] + ("..." if len(r) > MAX_STR_LEN else "")
+                    elif isinstance(r, list):
+                        if depth >= MAX_DEPTH:
+                            return "[Max depth reached]"
+                        return [trim_results(x, depth + 1) for x in r[:MAX_LIST_LEN]]
+                    elif isinstance(r, dict):
+                        if depth >= MAX_DEPTH:
+                            return "[Max depth reached]"
+                        trimmed = {}
+                        for i, (k, v) in enumerate(r.items()):
+                            if i >= MAX_DICT_KEYS:
+                                trimmed["_more_keys"] = f"... ({len(r) - MAX_DICT_KEYS} omitted)"
+                                break
+                            trimmed[k] = trim_results(v, depth + 1)
+                        return trimmed
+                    return r
+
+                for msg in user_ast_messages:
+                    role = msg.get("role")
+                    content = msg.get("content", "")
+                    msg_str = f"{role}: {content}"
+
+                    # Extract tool calls and chunks if this is an assistant message
+                    if role == "assistant":
+                        extracted_chunks = []
+
+                        # 1. From chunks list
+                        chunks = msg.get("chunks") or []
+                        if isinstance(chunks, list):
+                            last_tc = None
+                            for chunk in chunks:
+                                if isinstance(chunk, dict):
+                                    item = chunk.get("item", {})
+                                    if isinstance(item, dict) and item.get("type") in [
+                                        "tool_call",
+                                        "retrieval_call",
+                                    ]:
+                                        t_name = item.get("tool_name") or item.get("name") or "tool"
+                                        res = trim_results(item.get("results"))
+                                        tc = {"tool_name": t_name, "results": res}
+                                        extracted_chunks.append(tc)
+                                        last_tc = tc
+                                    elif chunk.get("type") in [
+                                        "response.tool_call.result",
+                                        "tool_call_result",
+                                    ]:
+                                        res = trim_results(chunk.get("result") or chunk)
+                                        if last_tc:
+                                            last_tc["results"] = res
+                                        else:
+                                            extracted_chunks.append(
+                                                {"tool_name": "tool", "results": res}
+                                            )
+
+                        # 2. From response_data dict/string
+                        resp_data = msg.get("response_data")
+                        if resp_data:
+                            if isinstance(resp_data, str):
+                                try:
+                                    resp_data = json.loads(resp_data)
+                                except Exception:
+                                    resp_data = {}
+                            if isinstance(resp_data, dict):
+                                t_calls = resp_data.get("tool_calls") or []
+                                if isinstance(t_calls, list):
+                                    for tc in t_calls:
+                                        if isinstance(tc, dict):
+                                            t_name = tc.get("name")
+                                            func = tc.get("function")
+                                            if isinstance(func, dict) and not t_name:
+                                                t_name = func.get("name")
+                                            res = trim_results(tc.get("result"))
+                                            extracted_chunks.append(
+                                                {"tool_name": t_name or "tool", "results": res}
+                                            )
+
+                        if extracted_chunks:
+                            chunks_strs = []
+                            for tc in extracted_chunks:
+                                t_name = tc.get("tool_name", "tool")
+                                res = tc.get("results")
+                                if res is not None:
+                                    res_str = json.dumps(res, ensure_ascii=False, default=str)
+                                    chunks_strs.append(f"[Tool: {t_name}, Results: {res_str}]")
+                            if chunks_strs:
+                                msg_str += "\nContext Chunks:\n" + "\n".join(chunks_strs)
+
+                    formatted_messages.append(msg_str)
+
+                prompt = "\n\n".join(formatted_messages)
 
         from agent import async_langflow_chat
 

tests/conftest.py

@@ -1,12 +1,19 @@
 import asyncio
 import os
+import sys
 import tempfile
 from pathlib import Path
 
 import pytest
 import pytest_asyncio
 from dotenv import load_dotenv
 
+# Ensure src is in sys.path
+ROOT = Path(__file__).resolve().parents[1]
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
 # Load environment variables
 load_dotenv()
 

tests/unit/test_chat_service_nudges.py

@@ -0,0 +1,128 @@
+"""Tests for ChatService langflow_nudges_chat extraction logic."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import agent
+from services.chat_service import ChatService
+
+
+@pytest.mark.asyncio
+async def test_langflow_nudges_chat_with_chunks_in_memory(monkeypatch):
+    """Ensure in-memory conversation chunks are formatted into the nudges prompt."""
+    monkeypatch.setattr("services.chat_service.LANGFLOW_URL", "http://localhost:7860")
+    monkeypatch.setattr("services.chat_service.NUDGES_FLOW_ID", "nudges-flow-id")
+
+    # Mock clients
+    mock_lf_client = MagicMock()
+    monkeypatch.setattr(
+        "services.chat_service.clients.ensure_langflow_client",
+        AsyncMock(return_value=mock_lf_client),
+    )
+
+    # Prepopulate active_conversations with a message containing tool call chunks
+    test_user_id = "test-user"
+    test_resp_id = "test-resp"
+
+    chunks_data = [
+        {
+            "item": {
+                "type": "tool_call",
+                "tool_name": "Retrieval",
+                "results": [{"filename": "doc.md", "text": "Important retrieved content."}],
+            }
+        }
+    ]
+
+    agent.active_conversations[test_user_id] = {
+        test_resp_id: {
+            "messages": [
+                {"role": "user", "content": "What is OpenRAG?"},
+                {"role": "assistant", "content": "It is an AI platform.", "chunks": chunks_data},
+            ]
+        }
+    }
+
+    # Mock async_langflow_chat to capture prompt
+    captured_prompt = None
+
+    async def mock_async_lf_chat(client, flow_id, prompt, user_id, **kwargs):
+        nonlocal captured_prompt
+        captured_prompt = prompt
+        return "Nudge 1\nNudge 2", "nudge-resp", []
+
+    monkeypatch.setattr(agent, "async_langflow_chat", mock_async_lf_chat)
+
+    svc = ChatService()
+    res = await svc.langflow_nudges_chat(user_id=test_user_id, previous_response_id=test_resp_id)
+
+    assert res["response"] == "Nudge 1\nNudge 2"
+    assert captured_prompt is not None
+    assert "user: What is OpenRAG?" in captured_prompt
+    assert "assistant: It is an AI platform." in captured_prompt
+    assert "Context Chunks:" in captured_prompt
+    assert "Important retrieved content." in captured_prompt
+
+    # Clean up
+    agent.active_conversations.pop(test_user_id, None)
+
+
+@pytest.mark.asyncio
+async def test_langflow_nudges_chat_with_chunks_langflow_history(monkeypatch):
+    """Ensure persistent Langflow history chunks are formatted into the nudges prompt."""
+    monkeypatch.setattr("services.chat_service.LANGFLOW_URL", "http://localhost:7860")
+    monkeypatch.setattr("services.chat_service.NUDGES_FLOW_ID", "nudges-flow-id")
+
+    # Mock clients
+    mock_lf_client = MagicMock()
+    monkeypatch.setattr(
+        "services.chat_service.clients.ensure_langflow_client",
+        AsyncMock(return_value=mock_lf_client),
+    )
+
+    # Ensure active_conversations is empty for this test
+    test_user_id = "lf-user"
+    test_resp_id = "lf-resp"
+    agent.active_conversations.pop(test_user_id, None)
+
+    # Mock langflow_history_service.get_session_messages
+    lf_messages = [
+        {"role": "user", "content": "Tell me about security."},
+        {
+            "role": "assistant",
+            "content": "Security is robust.",
+            "chunks": [
+                {
+                    "item": {
+                        "type": "tool_call",
+                        "tool_name": "OpenSearch",
+                        "results": {"hits": ["Security info chunk."]},
+                    }
+                }
+            ],
+        },
+    ]
+    mock_get_messages = AsyncMock(return_value=lf_messages)
+    from services.langflow_history_service import langflow_history_service
+
+    monkeypatch.setattr(langflow_history_service, "get_session_messages", mock_get_messages)
+
+    # Mock async_langflow_chat to capture prompt
+    captured_prompt = None
+
+    async def mock_async_lf_chat(client, flow_id, prompt, user_id, **kwargs):
+        nonlocal captured_prompt
+        captured_prompt = prompt
+        return "Nudge A\nNudge B", "nudge-resp", []
+
+    monkeypatch.setattr(agent, "async_langflow_chat", mock_async_lf_chat)
+
+    svc = ChatService()
+    res = await svc.langflow_nudges_chat(user_id=test_user_id, previous_response_id=test_resp_id)
+
+    assert res["response"] == "Nudge A\nNudge B"
+    assert captured_prompt is not None
+    assert "user: Tell me about security." in captured_prompt
+    assert "Context Chunks:" in captured_prompt
+    assert "Security info chunk." in captured_prompt