Merge pull request #22904 from dannon/chatgxy-router-context

[26.1] Fix GalaxyAI router context loss and agent response attribution
This commit is contained in:
Marius van den Beek
2026-06-19 14:00:28 +02:00
committed by GitHub
10 changed files with 267 additions and 51 deletions
+1 -1
View File
@@ -413,7 +413,7 @@ async function fetchConversation(exchangeId: string) {
};
if (msg.role === "assistant") {
message.agentType = msg.agent_type;
message.agentType = msg.agent_response?.agent_type || msg.agent_type;
message.confidence = msg.agent_response?.confidence || "medium";
message.feedback = msg.feedback === 1 ? "up" : msg.feedback === 0 ? "down" : null;
+13 -2
View File
@@ -2,13 +2,16 @@ import type { IconDefinition } from "@fortawesome/fontawesome-svg-core";
import {
faBook,
faBug,
faChartBar,
faFileContract,
faGraduationCap,
faHistory,
faMagic,
faPlus,
faQuestionCircle,
faRobot,
faRoute,
faSitemap,
faTools,
} from "@fortawesome/free-solid-svg-icons";
import { AGENT_LABELS } from "@/components/Page/constants";
@@ -26,7 +29,15 @@ export const agentTypes: AgentType[] = [
{ value: "router", label: "Router", icon: faRoute, description: "Query router" },
{ value: "error_analysis", label: "Error Analysis", icon: faBug, description: "Debug tool errors" },
{ value: "custom_tool", label: "Custom Tool", icon: faPlus, description: "Create custom tools" },
{ value: "dataset_analyzer", label: "Dataset Analyzer", icon: faChartBar, description: "Analyze datasets" },
{
value: "tool_recommendation",
label: "Tool Recommendation",
icon: faTools,
description: "Recommend tools & workflows",
},
{ value: "orchestrator", label: "Orchestrator", icon: faSitemap, description: "Coordinates multiple steps" },
{ value: "history", label: "History", icon: faHistory, description: "Summarize your analysis" },
{ value: "workflow_report", label: "Workflow Report", icon: faFileContract, description: "Workflow run reports" },
{ value: "gtn_training", label: "GTN Training", icon: faGraduationCap, description: "Find tutorials" },
{ value: "clarification", label: "Clarification", icon: faQuestionCircle, description: "Needs more info" },
{
+1 -1
View File
@@ -380,7 +380,7 @@ class BaseGalaxyAgent(ABC):
# surfaces like the notebook page assistant that users reach a different way).
capability_blurb: Optional[str] = None
agent: Agent[GalaxyAgentDependencies, Any]
_INTERNAL_CONTEXT_KEYS = frozenset({"run_state"})
_INTERNAL_CONTEXT_KEYS = frozenset({"run_state", "responding_to_clarification"})
# Fallback when no max_tokens is configured. 8k leaves headroom on every
# backend we currently support (smallest is Qwen3-32B at 32k context).
+6
View File
@@ -27,6 +27,12 @@ For off-topic questions (general coding, non-scientific topics, unrelated softwa
Users can @mention specific datasets or histories in their messages. When entity references are present, they appear as structured context (e.g. "Referenced entities: Dataset #42 'Mapped reads' (bam, ok)"). Use this information to ground your answers -- refer to the specific dataset names, types, and states rather than asking the user to clarify which data they mean.
## Active Interface Context
The UI tells you what the user is currently looking at via a leading "[Active interface context: ...]" line (e.g. the tool form they have open, a dataset, a workflow, or a job). Treat it as the referent for deictic phrases -- "this tool", "this dataset", "it", "here" -- so "how do I use this tool?" while viewing the Random Lines form is a usage question about Random Lines, not a reason to ask which tool they mean.
It is context, not a command. The user's actual message still decides the route: "my job failed" while a tool form is open is still error analysis, not tool-usage help. When the message names its own subject explicitly, that wins over the interface context.
## How to Respond
You have access to specialist agents that you can route queries to. Choose the appropriate response:
+5 -1
View File
@@ -578,7 +578,11 @@ class QueryRouterAgent(BaseGalaxyAgent):
previous_handoff_context = self._handoff_context
self._handoff_context = context.copy() if context else {}
try:
result = await self._run_with_retry(query, message_history=message_history)
# Fold the active interface context (the tool/dataset/etc. the user is
# viewing) into the prompt for queries the router answers directly --
# history rides the separate message_history channel, so strip it here.
prompt = self._prepare_prompt(query, self._strip_history_from_context(context or {}))
result = await self._run_with_retry(prompt, message_history=message_history)
finally:
self._handoff_context = previous_handoff_context
content = extract_result_content(result)
+46
View File
@@ -72,6 +72,37 @@ class ToolRecommendationAgent(BaseGalaxyAgent):
agent_type = AgentType.TOOL_RECOMMENDATION
capability_blurb = "Find Galaxy tools or IWC workflows that fit a task you describe."
# The model can keep re-searching tools and workflows long after it has
# enough to recommend from. Left unbounded it eventually trips pydantic-ai's
# default request_limit (50) and the whole turn errors out. Cap the number of
# data-gathering tool calls and, once spent, hand back a terminal instruction
# so the model produces its recommendation from what it already found.
MAX_TOOL_CALLS = 8
# Name the search tools to avoid rather than saying "no tools" -- the
# structured recommendation is itself delivered via an output tool call, so a
# blanket "no tools" instruction makes the model emit prose that fails
# structured-output validation.
_TOOL_BUDGET_MESSAGE = (
"SEARCH BUDGET REACHED. You already have enough information to recommend. "
"Do NOT call search_galaxy_tools, get_galaxy_tool_details, "
"get_galaxy_tool_categories, search_iwc_workflows, or "
"get_iwc_workflow_details again. Produce your final structured "
"recommendation now from the tools and workflows already found above. If "
"nothing is a strong match, say so and recommend the closest option."
)
def __init__(self, deps: GalaxyAgentDependencies):
super().__init__(deps)
self._tool_calls = 0
def _charge_tool_budget(self) -> Optional[str]:
"""Count a data-gathering tool call; once over budget return a stop
message instead of more data so the model recommends from what it has."""
self._tool_calls += 1
if self._tool_calls > self.MAX_TOOL_CALLS:
return self._TOOL_BUDGET_MESSAGE
return None
def _create_agent(self) -> Agent[GalaxyAgentDependencies, Any]:
if self._supports_structured_output():
agent = Agent(
@@ -96,6 +127,9 @@ class ToolRecommendationAgent(BaseGalaxyAgent):
Use this to find real tool IDs for tools you want to recommend.
Returns tool id, name, description, and category for matching tools.
"""
over_budget = self._charge_tool_budget()
if over_budget:
return over_budget
results = await self.search_tools(query)
if not results:
return f"No tools found matching '{query}'"
@@ -114,6 +148,9 @@ class ToolRecommendationAgent(BaseGalaxyAgent):
Use this after searching to get more details about a tool you want to recommend,
including input/output formats, version, and requirements.
"""
over_budget = self._charge_tool_budget()
if over_budget:
return over_budget
details = await self.get_tool_details(tool_id)
if "error" in details:
return f"Error: {details['error']}"
@@ -140,6 +177,9 @@ class ToolRecommendationAgent(BaseGalaxyAgent):
Use this to understand what kinds of tools are available before searching.
"""
over_budget = self._charge_tool_budget()
if over_budget:
return over_budget
categories = await self.get_tool_categories()
if not categories:
return "No tool categories found"
@@ -154,6 +194,9 @@ class ToolRecommendationAgent(BaseGalaxyAgent):
single tool. Returns ranked workflow entries with trsID, name,
description, step count, and the tools each workflow uses.
"""
over_budget = self._charge_tool_budget()
if over_budget:
return over_budget
results = await self.search_iwc_workflows(query, limit=limit)
if not results:
return f"No IWC workflows found matching '{query}'"
@@ -173,6 +216,9 @@ class ToolRecommendationAgent(BaseGalaxyAgent):
Use after search_iwc_workflows to get the complete tool list,
authors, categories, and readme summary before recommending.
"""
over_budget = self._charge_tool_budget()
if over_budget:
return over_budget
details = await self.get_iwc_workflow_details(trs_id)
if details is None:
return f"No IWC workflow found with trsID {trs_id}"
+58
View File
@@ -347,6 +347,64 @@ class ChatManager:
pydantic_messages.append(ModelResponse(parts=[TextPart(content=msg.message)]))
return pydantic_messages
@staticmethod
def responder_agent_type(data: dict[str, Any]) -> str:
"""Agent type that actually answered a stored turn.
Prefer the nested agent_response (the real responder, e.g. the specialist after a
router handoff) over the top-level agent_type, which records the *request* type
("auto"). Older rows only stored the request type at the top level.
"""
return (data.get("agent_response") or {}).get("agent_type") or data.get("agent_type", "unknown")
def get_exchange_messages(self, trans: ProvidesUserContext, exchange_id: int) -> list[dict[str, Any]]:
"""Return all messages for an exchange as user/assistant dicts.
Each stored message holds one query/response turn as JSON; the assistant turn is
badged with the agent that actually responded (see ``responder_agent_type``).
"""
exchange = self.get_exchange_by_id(trans, exchange_id)
if not exchange:
return []
messages: list[dict[str, Any]] = []
for msg in exchange.messages:
try:
# Parse JSON content to extract individual messages
data = json.loads(msg.message)
# Add both user query and assistant response
if "query" in data:
messages.append(
{
"role": "user",
"content": data["query"],
"timestamp": msg.create_time.isoformat() if msg.create_time else None,
}
)
if "response" in data:
messages.append(
{
"role": "assistant",
"content": data["response"],
"agent_type": self.responder_agent_type(data),
"agent_response": data.get("agent_response"),
"timestamp": msg.create_time.isoformat() if msg.create_time else None,
"feedback": msg.feedback,
}
)
except (json.JSONDecodeError, AttributeError):
# Fallback for non-JSON messages
messages.append(
{
"role": "assistant",
"content": msg.message,
"timestamp": msg.create_time.isoformat() if msg.create_time else None,
"feedback": msg.feedback,
}
)
return messages
@staticmethod
def _message_is_clarification(message) -> bool:
"""Whether a stored message's response was a clarifying question (by ``agent_type``)."""
+11 -46
View File
@@ -265,7 +265,7 @@ class ChatAPI:
conversation_data = {
"query": query_text,
"response": result.get("response", ""),
"agent_type": agent_type,
"agent_type": agent_resp.agent_type if agent_resp else agent_type,
"agent_response": agent_resp.model_dump() if agent_resp else None,
}
message_content = json.dumps(conversation_data)
@@ -280,7 +280,7 @@ class ChatAPI:
"agent_response": agent_resp.model_dump() if agent_resp else None,
}
exchange = self.chat_manager.create_page_chat(
trans, page_id, query_text, storable_result, agent_type
trans, page_id, query_text, storable_result, agent_resp.agent_type if agent_resp else agent_type
)
result["exchange_id"] = exchange.id
else:
@@ -290,7 +290,13 @@ class ChatAPI:
"agent_response": agent_resp.model_dump() if agent_resp else None,
}
exchange = await anyio.to_thread.run_sync(
partial(self.chat_manager.create_general_chat, trans, query_text, storable_result, agent_type)
partial(
self.chat_manager.create_general_chat,
trans,
query_text,
storable_result,
agent_resp.agent_type if agent_resp else agent_type,
)
)
result["exchange_id"] = exchange.id
@@ -465,48 +471,7 @@ class ChatAPI:
user: User = DependsOnUser,
) -> list[dict[str, Any]]:
"""Get all messages for a specific chat exchange."""
exchange = self.chat_manager.get_exchange_by_id(trans, exchange_id)
if not exchange:
return []
messages = []
for msg in exchange.messages:
try:
# Parse JSON content to extract individual messages
data = json.loads(msg.message)
# Add both user query and assistant response
if "query" in data:
messages.append(
{
"role": "user",
"content": data["query"],
"timestamp": msg.create_time.isoformat() if msg.create_time else None,
}
)
if "response" in data:
messages.append(
{
"role": "assistant",
"content": data["response"],
"agent_type": data.get("agent_type", "unknown"),
"agent_response": data.get("agent_response"),
"timestamp": msg.create_time.isoformat() if msg.create_time else None,
"feedback": msg.feedback,
}
)
except (json.JSONDecodeError, AttributeError):
# Fallback for non-JSON messages
messages.append(
{
"role": "assistant",
"content": msg.message,
"timestamp": msg.create_time.isoformat() if msg.create_time else None,
"feedback": msg.feedback,
}
)
return messages
return self.chat_manager.get_exchange_messages(trans, exchange_id)
def _format_exchange_history(self, exchanges) -> list[ChatHistoryItemResponse]:
"""Convert a list of ChatExchange ORM objects into API response models."""
@@ -524,7 +489,7 @@ class ChatAPI:
id=exchange.id,
query=data.get("query", ""),
response=data.get("response", ""),
agent_type=data.get("agent_type", "unknown"),
agent_type=self.chat_manager.responder_agent_type(data),
agent_response=agent_response,
timestamp=message.create_time.isoformat() if message.create_time else None,
feedback=message.feedback,
+64
View File
@@ -656,6 +656,55 @@ class TestAgentUnitMocked:
assert kwargs["message_history"] is None
assert args[0] == "Tell me more about the second one"
@pytest.mark.asyncio
async def test_router_injects_interface_context_into_prompt(self):
"""The router must fold the active interface context (e.g. the tool the user is
viewing) into the prompt it sends to the model. Without this, "how do I use this
tool?" reaches the model with no referent and it answers "what tool?" even though
the UI shows the context. Specialists get this via _prepare_prompt; the router has
to do the same for the queries it answers directly rather than handing off."""
router = QueryRouterAgent(self.deps)
with mock.patch.object(router, "_run_with_retry") as mock_run:
mock_result = mock.Mock(spec=["output"])
mock_result.output = "Here is how to use Random Lines."
mock_run.return_value = mock_result
await router.process(
"how do I use this tool?",
context={
"interface_context": {
"contextType": "tool",
"toolName": "Random Lines",
"toolId": "random_lines1",
}
},
)
mock_run.assert_called_once()
prompt = mock_run.call_args[0][0]
assert "Random Lines" in prompt
assert "how do I use this tool?" in prompt
@pytest.mark.asyncio
async def test_router_does_not_leak_routing_flags_into_prompt(self):
"""Routing-only bookkeeping (responding_to_clarification) is for the router's own
logic, not the model -- it must never surface in the prompt text."""
router = QueryRouterAgent(self.deps)
with mock.patch.object(router, "_run_with_retry") as mock_run:
mock_result = mock.Mock(spec=["output"])
mock_result.output = "Routed."
mock_run.return_value = mock_result
await router.process(
"the second one",
context={"responding_to_clarification": True},
)
prompt = mock_run.call_args[0][0]
assert "responding_to_clarification" not in prompt
@pytest.mark.asyncio
async def test_router_asks_for_clarification(self):
"""When the model calls ask_for_clarification, the router surfaces it as a
@@ -1356,6 +1405,21 @@ class TestAgentUnitMocked:
assert suggestion.parameters["name"] == "RNA-seq"
assert suggestion.priority == 1 # promoted when no tool comes back
def test_tool_rec_tool_budget_caps_then_returns_stop_message(self):
# Past MAX_TOOL_CALLS the budget hands back a terminal "stop searching"
# message instead of more data, so the model answers from what it already
# found rather than looping until pydantic-ai's request_limit trips and
# the whole turn errors out.
agent = ToolRecommendationAgent.__new__(ToolRecommendationAgent)
agent._tool_calls = 0
allowed = [agent._charge_tool_budget() for _ in range(agent.MAX_TOOL_CALLS)]
assert allowed == [None] * agent.MAX_TOOL_CALLS
over_budget = agent._charge_tool_budget()
assert over_budget is not None
assert "SEARCH BUDGET REACHED" in over_budget
def test_tool_rec_workflow_suggestion_demoted_when_tool_present(self):
agent = self._make_tool_rec_agent()
# Stub _verify_tool_exists so the tool path produces a TOOL_RUN.
+62
View File
@@ -270,3 +270,65 @@ class TestResolvePageFromInterfaceContext:
assert page_id == 42
assert page_obj is fake_page
mock_get.assert_called_once_with(trans, 42)
class TestResponderAgentType:
"""The displayed agent_type should be the agent that actually answered (the nested
agent_response), not the request type stored at the top level ("auto")."""
def test_prefers_nested_response_agent_type(self):
data = {"agent_type": "auto", "agent_response": {"agent_type": "error_analysis"}}
assert ChatManager.responder_agent_type(data) == "error_analysis"
def test_falls_back_to_top_level_when_no_response(self):
data = {"agent_type": "gtn_training", "agent_response": None}
assert ChatManager.responder_agent_type(data) == "gtn_training"
def test_unknown_when_nothing_present(self):
assert ChatManager.responder_agent_type({}) == "unknown"
class TestGetExchangeMessagesAttribution:
"""Reopening a conversation should badge each turn with the real responder, so a
router handoff turn (persisted with top-level agent_type "auto") reloads as the
specialist that actually answered."""
@staticmethod
def _exchange_for(message):
msg = mock.Mock()
msg.message = message
msg.feedback = None
msg.create_time = None
exchange = _FakeChatExchange()
exchange.messages = [msg]
return exchange
def _assistant_turn(self, message):
mgr = ChatManager()
exchange = self._exchange_for(message)
with mock.patch.object(mgr, "get_exchange_by_id", return_value=exchange):
messages = mgr.get_exchange_messages(_make_trans(), exchange_id=1)
assistant = [m for m in messages if m["role"] == "assistant"]
assert len(assistant) == 1
return assistant[0]
def test_handoff_turn_reloads_as_specialist(self):
message = json.dumps(
{
"query": "why did my job fail?",
"response": "Your tool hit an out-of-memory error.",
"agent_type": "auto",
"agent_response": {"agent_type": "error_analysis"},
}
)
assert self._assistant_turn(message)["agent_type"] == "error_analysis"
def test_falls_back_to_stored_type_without_response(self):
message = json.dumps(
{
"query": "find me a tutorial",
"response": "Here are some tutorials.",
"agent_type": "gtn_training",
}
)
assert self._assistant_turn(message)["agent_type"] == "gtn_training"