Jodinho commited on
Commit
ed8a277
·
1 Parent(s): 311871d

fix(router): architectural fix for multi-turn conversational continuations

Browse files

- Pass recent history to classify_query for context-aware routing
- Add multi-turn continuation rules to ROUTER_PROMPT
- Add deterministic _rescue_classification guardrail for active sessions
- Increase max_tokens in router (1500) and agent loop (3500) to prevent thinking tag cutoffs
- Add unit test coverage for multi-turn and continuation classification

services/agent_loop.py CHANGED
@@ -115,7 +115,7 @@ async def resolve_suggested_actions(user_query: str, assistant_reply: str) -> li
115
  messages=messages,
116
  tools=[SUGGEST_ACTIONS_TOOL],
117
  tool_choice={"type": "function", "function": {"name": "suggest_actions"}},
118
- max_tokens=250,
119
  temperature=0.0,
120
  )
121
  actions = _extract_suggested_actions_from_response(res)
@@ -157,7 +157,7 @@ async def _call_retrieval_model(messages: list[dict], active_tools: list, round_
157
  messages=messages,
158
  tools=active_tools,
159
  tool_choice="auto",
160
- max_tokens=2500,
161
  temperature=0.2,
162
  )
163
  return res.choices[0].message
@@ -195,7 +195,7 @@ async def _synthesize_markdown(messages: list[dict]) -> str:
195
  messages=synth_messages,
196
  tools=None,
197
  tool_choice="none",
198
- max_tokens=2500,
199
  temperature=0.2,
200
  )
201
  raw = res.choices[0].message.content or ""
@@ -252,7 +252,7 @@ async def _stream_synthesis(synth_messages: list[dict], loop: asyncio.AbstractEv
252
  tools=None,
253
  tool_choice="none",
254
  temperature=0.2,
255
- max_tokens=2500,
256
  stream=True,
257
  ),
258
  )
@@ -389,7 +389,8 @@ async def process_chat_message(
389
  with propagate_attributes(session_id=session_id, tags=["real-estate-chat"]):
390
  get_client().update_current_span(input=user_query)
391
 
392
- classification, tool_categories = await classify_query(user_query)
 
393
 
394
  if classification == "OUT_OF_SCOPE":
395
  get_client().update_current_span(output=_REPLY_OUT_OF_SCOPE)
@@ -437,7 +438,8 @@ async def stream_chat_message(
437
  ) -> AsyncIterator[dict]:
438
  """SSE streaming chat handler."""
439
  with propagate_attributes(session_id=session_id, tags=["real-estate-chat", "stream"]):
440
- classification, tool_categories = await classify_query(user_query)
 
441
  yield {"type": "status", "classification": classification}
442
 
443
  if classification == "OUT_OF_SCOPE":
 
115
  messages=messages,
116
  tools=[SUGGEST_ACTIONS_TOOL],
117
  tool_choice={"type": "function", "function": {"name": "suggest_actions"}},
118
+ max_tokens=500,
119
  temperature=0.0,
120
  )
121
  actions = _extract_suggested_actions_from_response(res)
 
157
  messages=messages,
158
  tools=active_tools,
159
  tool_choice="auto",
160
+ max_tokens=3500,
161
  temperature=0.2,
162
  )
163
  return res.choices[0].message
 
195
  messages=synth_messages,
196
  tools=None,
197
  tool_choice="none",
198
+ max_tokens=3500,
199
  temperature=0.2,
200
  )
201
  raw = res.choices[0].message.content or ""
 
252
  tools=None,
253
  tool_choice="none",
254
  temperature=0.2,
255
+ max_tokens=3500,
256
  stream=True,
257
  ),
258
  )
 
389
  with propagate_attributes(session_id=session_id, tags=["real-estate-chat"]):
390
  get_client().update_current_span(input=user_query)
391
 
392
+ recent_history = get_context_window(session_id, limit=4)
393
+ classification, tool_categories = await classify_query(user_query, recent_history=recent_history)
394
 
395
  if classification == "OUT_OF_SCOPE":
396
  get_client().update_current_span(output=_REPLY_OUT_OF_SCOPE)
 
438
  ) -> AsyncIterator[dict]:
439
  """SSE streaming chat handler."""
440
  with propagate_attributes(session_id=session_id, tags=["real-estate-chat", "stream"]):
441
+ recent_history = get_context_window(session_id, limit=4)
442
+ classification, tool_categories = await classify_query(user_query, recent_history=recent_history)
443
  yield {"type": "status", "classification": classification}
444
 
445
  if classification == "OUT_OF_SCOPE":
services/prompts.py CHANGED
@@ -3,15 +3,22 @@ Prompt definitions and dynamic prompt composer for the Joule Dynamics Real Estat
3
  """
4
 
5
  ROUTER_PROMPT = """You are the classification router for the Joule Dynamics Real Estate Intelligence Layer.
6
- Analyze the user query and classify it into EXACTLY ONE of seven classifications:
7
-
8
- 1. "OUT_OF_SCOPE": Query asks about topics completely unrelated to real estate, data monitoring, or the current conversation (e.g. sports, general coding, cooking, recipes). IMPORTANT: Questions asking who you are ("who are you?"), what model you are, or asking to summarize/recap what was discussed in the current chat ("what have we discussed so far?", "summarize our chat") are ALWAYS IN-SCOPE and must NEVER be classified as OUT_OF_SCOPE.
9
- 2. "PATH_A": Query asks a live-data question (prices, spikes, availability, market averages, KPIs, specific listing rates).
 
 
 
 
 
 
10
  3. "PATH_B": Query asks a dashboard visual/UI question (e.g. "What do the top 4 metric cards mean?", "Why does the table say 'Was Available'?", "What do the green/red map pins mean?", "Why is the chart line dotted?", "What does the sparkline mean?", "Why is a row dimmed?"), OR asks a real estate methodology/business logic question (7-day average definition, 2-night check-in window, 4x daily scrape cadence, booked vs host-blocked, temporal UX states), OR asks a meta-conversational question (e.g. "What have we discussed so far?", "Can you summarize our conversation?", "Recap what we talked about").
11
  4. "BOTH": Query requires BOTH explaining a dashboard UI/methodology concept AND fetching live data metrics via tools.
12
  5. "GREETING": User is saying hello, thanking the assistant, asking "who are you?", or making casual conversation.
13
  6. "COMMERCIAL_HANDOFF": Query asks about getting started, hiring Joule Dynamics, custom builds, custom dashboards, pricing for software, or requests tracking for their own specific portfolio outside the demo scope.
14
- 7. "ALERT_SUBSCRIPTION": User wants to be notified about a future event — price spikes, rate changes, availability changes, new listings, pricing anomalies, volatility, trend reversals, or recurring market digests. Keywords: "alert me", "notify me", "let me know when", "tell me when", "tell me whenever", "subscribe", "watch this", "track this for me". This is DISTINCT from PATH_A (which asks about current/historical data, not future notifications).
 
15
 
16
  Respond ONLY with valid JSON matching this schema:
17
  {
 
3
  """
4
 
5
  ROUTER_PROMPT = """You are the classification router for the Joule Dynamics Real Estate Intelligence Layer.
6
+ Analyze the user's latest query in the context of the ongoing conversation and classify it into EXACTLY ONE of seven classifications:
7
+
8
+ 1. "OUT_OF_SCOPE": Query asks about topics completely unrelated to real estate, data monitoring, or the current conversation (e.g. sports, general coding, cooking, recipes).
9
+ CRITICAL MULTI-TURN CONTINUATION RULES:
10
+ - If the conversation is already in progress and the user is answering or continuing an ongoing topic, it is NEVER OUT_OF_SCOPE.
11
+ - An answer providing an email address, frequency (daily/weekly), threshold percentage, or confirmation for an alert setup MUST be classified as "ALERT_SUBSCRIPTION".
12
+ - An answer specifying a market name (e.g. "Abuja", "Miami"), property name, date range, or filter in response to an assistant question MUST be classified as "PATH_A" or "BOTH".
13
+ - Questions asking who you are ("who are you?"), what model you are, or asking to summarize/recap what was discussed ("what have we discussed so far?", "summarize our chat") are ALWAYS IN-SCOPE (classify as "GREETING" or "PATH_B").
14
+ - ONLY classify as "OUT_OF_SCOPE" if the user explicitly switches the topic to a completely unrelated subject outside real estate and outside the chat history.
15
+ 2. "PATH_A": Query asks a live-data question (prices, spikes, availability, market averages, KPIs, specific listing rates), OR provides clarification (market, date, property) for an active data query.
16
  3. "PATH_B": Query asks a dashboard visual/UI question (e.g. "What do the top 4 metric cards mean?", "Why does the table say 'Was Available'?", "What do the green/red map pins mean?", "Why is the chart line dotted?", "What does the sparkline mean?", "Why is a row dimmed?"), OR asks a real estate methodology/business logic question (7-day average definition, 2-night check-in window, 4x daily scrape cadence, booked vs host-blocked, temporal UX states), OR asks a meta-conversational question (e.g. "What have we discussed so far?", "Can you summarize our conversation?", "Recap what we talked about").
17
  4. "BOTH": Query requires BOTH explaining a dashboard UI/methodology concept AND fetching live data metrics via tools.
18
  5. "GREETING": User is saying hello, thanking the assistant, asking "who are you?", or making casual conversation.
19
  6. "COMMERCIAL_HANDOFF": Query asks about getting started, hiring Joule Dynamics, custom builds, custom dashboards, pricing for software, or requests tracking for their own specific portfolio outside the demo scope.
20
+ 7. "ALERT_SUBSCRIPTION": User wants to be notified about a future event — price spikes, rate changes, availability changes, new listings, pricing anomalies, volatility, trend reversals, or recurring market digests (e.g. "alert me", "notify me", "let me know when", "tell me when", "tell me whenever", "subscribe", "watch this", "track this for me").
21
+ CRITICAL: ALSO includes any follow-up response where the user provides an email address, threshold, frequency, or market choice after the assistant offered or asked to set up an alert!
22
 
23
  Respond ONLY with valid JSON matching this schema:
24
  {
services/query_router.py CHANGED
@@ -53,19 +53,71 @@ def _normalize_tool_categories(categories: object) -> list[str]:
53
  return []
54
 
55
 
56
- async def classify_query(user_query: str) -> tuple[str, list[str]]:
57
- """Classify user query and extract tool categories for category gating."""
58
- messages = [
59
- {"role": "system", "content": ROUTER_PROMPT},
60
- {"role": "user", "content": user_query},
61
- ]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
62
 
63
  for model in [GROQ_ROUTE_MODEL, GROQ_FALLBACK_ROUTE_MODEL]:
64
  try:
65
  res = await groq_call(
66
  model=model,
67
  messages=messages,
68
- max_tokens=600,
69
  temperature=0.0,
70
  )
71
  raw = res.choices[0].message.content or "{}"
@@ -79,6 +131,8 @@ async def classify_query(user_query: str) -> tuple[str, list[str]]:
79
  )
80
  classification = "PATH_A"
81
 
 
 
82
  logger.info(
83
  f"[router] {model} -> {classification} (categories={tool_categories}): {data.get('reason', '')}"
84
  )
@@ -90,4 +144,5 @@ async def classify_query(user_query: str) -> tuple[str, list[str]]:
90
  else:
91
  logger.warning(f"[router] {model} failed, trying fallback: {exc}")
92
 
93
- return "PATH_A", []
 
 
53
  return []
54
 
55
 
56
+ def _build_router_messages(user_query: str, recent_history: list[dict] | None = None) -> list[dict]:
57
+ """Constructs router message list with recent conversation history for multi-turn awareness."""
58
+ messages = [{"role": "system", "content": ROUTER_PROMPT}]
59
+ if recent_history:
60
+ for msg in recent_history[-2:]:
61
+ content = msg.get("content", "")
62
+ if "User Query:" in content:
63
+ content = content.split("User Query:", 1)[-1].strip()
64
+ messages.append({
65
+ "role": msg.get("role", "user"),
66
+ "content": content[:400],
67
+ })
68
+ messages.append({"role": "user", "content": user_query})
69
+ return messages
70
+
71
+
72
+ def _find_last_assistant_message(recent_history: list[dict]) -> str:
73
+ """Extracts the text of the most recent assistant turn in the history."""
74
+ for msg in reversed(recent_history):
75
+ if msg.get("role") == "assistant":
76
+ return msg.get("content", "").lower()
77
+ return ""
78
+
79
+
80
+ def _rescue_classification(
81
+ classification: str,
82
+ user_query: str,
83
+ recent_history: list[dict] | None,
84
+ ) -> str:
85
+ """Architectural guardrail: rescues false OUT_OF_SCOPE classifications during active conversational flows."""
86
+ if classification != "OUT_OF_SCOPE" or not recent_history:
87
+ return classification
88
+
89
+ query_lower = user_query.strip().lower()
90
+
91
+ if re.search(r"[\w\.-]+@[\w\.-]+\.\w+", query_lower):
92
+ logger.info("[router] Rescuing OUT_OF_SCOPE -> ALERT_SUBSCRIPTION (email address in active session)")
93
+ return "ALERT_SUBSCRIPTION"
94
+
95
+ last_assistant_msg = _find_last_assistant_message(recent_history)
96
+ alert_keywords = ("alert", "notification", "email address", "subscribe", "digest")
97
+ if any(kw in last_assistant_msg for kw in alert_keywords) and len(query_lower.split()) <= 8:
98
+ logger.info("[router] Rescuing OUT_OF_SCOPE -> ALERT_SUBSCRIPTION (continuation of alert setup)")
99
+ return "ALERT_SUBSCRIPTION"
100
+
101
+ if len(query_lower.split()) <= 5 and any(c in last_assistant_msg for c in ("?", "which", "would you", "what")):
102
+ logger.info("[router] Rescuing OUT_OF_SCOPE -> PATH_A (short answer to assistant question)")
103
+ return "PATH_A"
104
+
105
+ return classification
106
+
107
+
108
+ async def classify_query(
109
+ user_query: str,
110
+ recent_history: list[dict] | None = None,
111
+ ) -> tuple[str, list[str]]:
112
+ """Classify user query and extract tool categories for category gating with multi-turn awareness."""
113
+ messages = _build_router_messages(user_query, recent_history)
114
 
115
  for model in [GROQ_ROUTE_MODEL, GROQ_FALLBACK_ROUTE_MODEL]:
116
  try:
117
  res = await groq_call(
118
  model=model,
119
  messages=messages,
120
+ max_tokens=1500,
121
  temperature=0.0,
122
  )
123
  raw = res.choices[0].message.content or "{}"
 
131
  )
132
  classification = "PATH_A"
133
 
134
+ classification = _rescue_classification(classification, user_query, recent_history)
135
+
136
  logger.info(
137
  f"[router] {model} -> {classification} (categories={tool_categories}): {data.get('reason', '')}"
138
  )
 
144
  else:
145
  logger.warning(f"[router] {model} failed, trying fallback: {exc}")
146
 
147
+ fallback_classification = _rescue_classification("PATH_A", user_query, recent_history)
148
+ return fallback_classification, []
tests/test_alert_system.py CHANGED
@@ -37,6 +37,51 @@ class TestRouterClassification:
37
  from services.query_router import _VALID_CLASSIFICATIONS
38
  assert len(_VALID_CLASSIFICATIONS) == 7
39
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
40
 
41
  class TestToolResolution:
42
  """Verify _resolve_active_tools returns exactly 2 tools for ALERT_SUBSCRIPTION."""
 
37
  from services.query_router import _VALID_CLASSIFICATIONS
38
  assert len(_VALID_CLASSIFICATIONS) == 7
39
 
40
+ def test_rescue_classification_email_address(self):
41
+ from services.query_router import _rescue_classification
42
+ history = [
43
+ {"role": "user", "content": "Send me a report whenever rate changes in Abuja"},
44
+ {"role": "assistant", "content": "Sure thing! To set up a notification for any rate changes in Abuja, I'll need the email address where you'd like to receive the alerts."},
45
+ ]
46
+ result = _rescue_classification("OUT_OF_SCOPE", "john.albarka.ibrahim@gmail.com", history)
47
+ assert result == "ALERT_SUBSCRIPTION"
48
+
49
+ def test_rescue_classification_alert_short_reply(self):
50
+ from services.query_router import _rescue_classification
51
+ history = [
52
+ {"role": "assistant", "content": "Would you like a daily or weekly digest for Miami?"},
53
+ ]
54
+ assert _rescue_classification("OUT_OF_SCOPE", "weekly", history) == "ALERT_SUBSCRIPTION"
55
+ assert _rescue_classification("OUT_OF_SCOPE", "daily please", history) == "ALERT_SUBSCRIPTION"
56
+
57
+ def test_rescue_classification_general_question_short_reply(self):
58
+ from services.query_router import _rescue_classification
59
+ history = [
60
+ {"role": "assistant", "content": "Which market would you like me to analyze?"},
61
+ ]
62
+ assert _rescue_classification("OUT_OF_SCOPE", "Miami", history) == "PATH_A"
63
+
64
+ def test_rescue_classification_truly_out_of_scope_preserved(self):
65
+ from services.query_router import _rescue_classification
66
+ history = [
67
+ {"role": "assistant", "content": "Sure thing! I need your email address."},
68
+ ]
69
+ result = _rescue_classification("OUT_OF_SCOPE", "how do I bake a chocolate cake with vanilla frosting?", history)
70
+ assert result == "OUT_OF_SCOPE"
71
+
72
+ def test_build_router_messages_includes_history(self):
73
+ from services.query_router import _build_router_messages
74
+ history = [
75
+ {"role": "user", "content": "User Context Filters: {}\nUser Query: Track Abuja"},
76
+ {"role": "assistant", "content": "What is your email address?"},
77
+ ]
78
+ messages = _build_router_messages("john@example.com", history)
79
+ assert len(messages) == 4
80
+ assert messages[0]["role"] == "system"
81
+ assert messages[1]["content"] == "Track Abuja"
82
+ assert messages[2]["content"] == "What is your email address?"
83
+ assert messages[3]["content"] == "john@example.com"
84
+
85
 
86
  class TestToolResolution:
87
  """Verify _resolve_active_tools returns exactly 2 tools for ALERT_SUBSCRIPTION."""