Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 30 additions & 22 deletions configs/config_frontier_models.yml
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
# This is an example configuration for hybrid frontier model mode.
# This is an example configuration for GPT frontier model mode.
# It has the following features:
# - A combination of a open-source model for researcher and a frontier model for orchestrating
# - GPT Sol for orchestration, planning, and report writing
# - GPT Luna for source routing and research
# - OpenAI API access through the standard OPENAI_API_KEY
# - Knowledge retrieval using LlamaIndex
# - Web search and Paper search tools by default

Expand Down Expand Up @@ -70,23 +72,29 @@ llms:
chat_template_kwargs:
enable_thinking: true

nemotron_ultra_llm:
_type: nim
model_name: nvidia/nemotron-3-ultra-550b-a55b
base_url: "https://integrate.api.nvidia.com/v1"
temperature: 0.7
top_p: 0.7
max_tokens: 65536
num_retries: 5
chat_template_kwargs:
enable_thinking: true
gpt_sol_agent_llm:
_type: openai
model_name: "gpt-5.6-sol"
api_key: ${OPENAI_API_KEY}
max_tokens: 16384
num_retries: 2
parallel_tool_calls: false

gpt_sol_writer_llm:
_type: openai
model_name: "gpt-5.6-sol"
api_key: ${OPENAI_API_KEY}
max_tokens: 32768
num_retries: 2
parallel_tool_calls: false
Comment thread
AjayThorve marked this conversation as resolved.
Resolved

# Frontier model for orchestrating and planning in deep research agent
# You need to set the OPENAI_API_KEY environment variable
openai_gpt_5_2:
gpt_luna_agent_llm:
_type: openai
model_name: "gpt-5.2"
model_name: "gpt-5.6-luna"
api_key: ${OPENAI_API_KEY}
max_tokens: 16384
num_retries: 2
parallel_tool_calls: false

# LLM for document summaries (required when generate_summary: true)
summary_llm:
Expand Down Expand Up @@ -137,7 +145,7 @@ functions:

clarifier_agent:
_type: clarifier_agent
llm: openai_gpt_5_2
llm: gpt_sol_agent_llm
tools:
- web_search_tool
- knowledge_search
Expand All @@ -158,11 +166,11 @@ functions:
deep_research_agent:
_type: deep_research_agent
enable_citation_verification: true
orchestrator_llm: openai_gpt_5_2
source_router_llm: nemotron_ultra_llm
researcher_llm: nemotron_ultra_llm
planner_llm: openai_gpt_5_2
writer_llm: openai_gpt_5_2
orchestrator_llm: gpt_sol_agent_llm
source_router_llm: gpt_luna_agent_llm
researcher_llm: gpt_luna_agent_llm
planner_llm: gpt_sol_agent_llm
writer_llm: gpt_sol_writer_llm
verbose: true
tools:
# - paper_search_tool # Uncomment if SERPER_API_KEY is set
Expand Down
59 changes: 59 additions & 0 deletions src/aiq_agent/agents/deep_researcher/custom_middleware.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@
from langchain_core.messages import HumanMessage
from langchain_core.messages import SystemMessage
from langchain_core.messages import ToolMessage
from pydantic import BaseModel
from pydantic import ValidationError

from aiq_agent.common import get_source_id_for_tool
from aiq_agent.common import load_prompt
Expand Down Expand Up @@ -98,6 +100,63 @@ def _tool_result_failed(result: object) -> bool:
return status == "error"


class StructuredResponseTextFallbackMiddleware(AgentMiddleware):
"""Recover one exact JSON response when a provider skips the output tool."""

def __init__(self, schema: type[BaseModel]) -> None:
self.schema = schema
schema_json = json.dumps(schema.model_json_schema(), separators=(",", ":"), ensure_ascii=False)
self._correction = (
"The previous response did not produce the required structured result. Do not call tools. "
"Return exactly one JSON object matching this JSON Schema, with no Markdown fences or prose:\n"
f"{schema_json}"
)

def _promote(self, response: ModelResponse) -> ModelResponse:
if response.structured_response is not None or len(response.result) != 1:
return response
message = response.result[0]
if not isinstance(message, AIMessage) or message.tool_calls or not isinstance(message.content, str):
return response
try:
structured = self.schema.model_validate_json(message.content)
except ValidationError:
return response
logger.info("Recovered %s from schema-valid JSON message content", self.schema.__name__)
return ModelResponse(result=response.result, structured_response=structured)

@staticmethod
def _needs_correction(response: ModelResponse) -> bool:
if response.structured_response is not None or len(response.result) != 1:
return False
message = response.result[0]
return isinstance(message, AIMessage) and not message.tool_calls and isinstance(message.content, str)

def _correction_request(self, request):
return request.override(
messages=[*request.messages, HumanMessage(content=self._correction)],
tools=[],
tool_choice=None,
response_format=None,
)

def wrap_model_call(self, request, handler):
"""Promote JSON text, with one tools-disabled corrective call when needed."""
response = self._promote(handler(request))
if not self._needs_correction(response):
return response
logger.warning("Retrying %s as a tools-disabled JSON response", self.schema.__name__)
return self._promote(handler(self._correction_request(request)))

async def awrap_model_call(self, request, handler):
"""Promote JSON text, with one tools-disabled corrective call when needed."""
response = self._promote(await handler(request))
if not self._needs_correction(response):
return response
logger.warning("Retrying %s as a tools-disabled JSON response", self.schema.__name__)
return self._promote(await handler(self._correction_request(request)))


class FinalReportCommitTracker:
"""Run-local proof of the writer's most recent successful report mutation."""

Expand Down
4 changes: 4 additions & 0 deletions src/aiq_agent/agents/deep_researcher/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
from .custom_middleware import SourceRoutingGuardMiddleware
from .custom_middleware import SourceRoutingPersistenceMiddleware
from .custom_middleware import StateMutationGuardMiddleware
from .custom_middleware import StructuredResponseTextFallbackMiddleware
from .custom_middleware import TodoQuotaMiddleware
from .custom_middleware import TodoSuppressionMiddleware
from .custom_middleware import ToolNameSanitizationMiddleware
Expand Down Expand Up @@ -386,6 +387,7 @@ def build_researcher_runnable(
FilesystemMiddleware(backend=backend, _permissions=filesystem_permissions),
create_summarization_middleware(researcher_model, backend),
PatchToolCallsMiddleware(),
StructuredResponseTextFallbackMiddleware(ResearchNotes),
*researcher_middleware,
*(visibility_middleware or []),
]
Expand Down Expand Up @@ -457,6 +459,7 @@ def build_deep_research_subagents(context: DeepResearchGraphContext) -> list[dic
sandbox_enabled=context.runtime.execution_enabled,
),
TodoSuppressionMiddleware(),
StructuredResponseTextFallbackMiddleware(SourceRoutingPlan),
SourceRoutingPersistenceMiddleware(
backend=context.backend,
state_budget=context.state_budget,
Expand Down Expand Up @@ -487,6 +490,7 @@ def build_deep_research_subagents(context: DeepResearchGraphContext) -> list[dic
sandbox_enabled=context.runtime.execution_enabled,
),
TodoSuppressionMiddleware(),
StructuredResponseTextFallbackMiddleware(ResearchPlan),
PlanPersistenceMiddleware(
backend=context.backend,
state_budget=context.state_budget,
Expand Down
126 changes: 126 additions & 0 deletions tests/aiq_agent/agents/deep_researcher/test_custom_middleware.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
from deepagents.backends import StateBackend
from deepagents.middleware.filesystem import FilesystemMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware.types import ModelResponse
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from langchain_core.messages import AIMessage
from langchain_core.messages import HumanMessage
Expand All @@ -44,10 +45,12 @@
from aiq_agent.agents.deep_researcher.custom_middleware import SourceRoutingGuardMiddleware
from aiq_agent.agents.deep_researcher.custom_middleware import SourceRoutingPersistenceMiddleware
from aiq_agent.agents.deep_researcher.custom_middleware import StateMutationGuardMiddleware
from aiq_agent.agents.deep_researcher.custom_middleware import StructuredResponseTextFallbackMiddleware
from aiq_agent.agents.deep_researcher.custom_middleware import TodoQuotaMiddleware
from aiq_agent.agents.deep_researcher.custom_middleware import TodoSuppressionMiddleware
from aiq_agent.agents.deep_researcher.custom_middleware import ToolNameSanitizationMiddleware
from aiq_agent.agents.deep_researcher.custom_middleware import ToolVisibilityMiddleware
from aiq_agent.agents.deep_researcher.models import SourceRoutingPlan
from aiq_agent.agents.deep_researcher.resource_limits import DeepResearchResourceLimits
from aiq_agent.agents.deep_researcher.resource_limits import StateBudgetLedger
from aiq_agent.agents.deep_researcher.tools.source_registry import build_get_verified_sources_tool
Expand All @@ -63,6 +66,129 @@ def bind_tools(self, tools, *, tool_choice=None, **kwargs):
return self


class TestStructuredResponseTextFallbackMiddleware:
"""Strictly recover provider responses that contain the requested contract as JSON text."""

@staticmethod
def _routing() -> dict[str, object]:
return {
"domain_id": "general",
"domain_name": "General",
"routing_reason": "Best fit",
"recommendations": [],
"fallback_sources": [],
"planner_guidance": "Use web search.",
}

def test_promotes_exact_schema_valid_json_in_agent(self) -> None:
payload = self._routing()
model = _ToolBindingFakeChatModel(responses=[AIMessage(content=json.dumps(payload))])
agent = create_agent(
model=model,
tools=[],
middleware=[StructuredResponseTextFallbackMiddleware(SourceRoutingPlan)],
response_format=SourceRoutingPlan,
)

result = agent.invoke({"messages": [HumanMessage(content="Route this request.")]})

assert result["structured_response"] == SourceRoutingPlan.model_validate(payload)

def test_corrects_empty_response_in_agent(self) -> None:
payload = self._routing()
model = _ToolBindingFakeChatModel(responses=[AIMessage(content=""), AIMessage(content=json.dumps(payload))])
agent = create_agent(
model=model,
tools=[],
middleware=[StructuredResponseTextFallbackMiddleware(SourceRoutingPlan)],
response_format=SourceRoutingPlan,
)

result = agent.invoke({"messages": [HumanMessage(content="Route this request.")]})

assert result["structured_response"] == SourceRoutingPlan.model_validate(payload)

@pytest.mark.parametrize(
"content",
["", "```json\n{}\n```", "{}\nUse this routing plan.", "[]", "{}"],
)
def test_retries_non_exact_or_schema_invalid_content_without_tools(self, content: str) -> None:
payload = self._routing()
responses = [
ModelResponse(result=[AIMessage(content=content)]),
ModelResponse(result=[AIMessage(content=json.dumps(payload))]),
]
request = MagicMock()
request.messages = [HumanMessage(content="Route this request.")]
corrected_request = object()
request.override.return_value = corrected_request
handler = MagicMock(side_effect=responses)
middleware = StructuredResponseTextFallbackMiddleware(SourceRoutingPlan)

result = middleware.wrap_model_call(request, handler)

assert result.structured_response == SourceRoutingPlan.model_validate(payload)
assert handler.call_args_list == [((request,),), ((corrected_request,),)]
request.override.assert_called_once()
overrides = request.override.call_args.kwargs
assert overrides["tools"] == []
assert overrides["tool_choice"] is None
assert overrides["response_format"] is None
assert overrides["messages"][:-1] == request.messages
correction = overrides["messages"][-1]
assert isinstance(correction, HumanMessage)
assert "exactly one JSON object" in str(correction.content)
assert '"domain_id"' in str(correction.content)

@pytest.mark.asyncio
async def test_async_correction_is_bounded_to_one_retry(self) -> None:
responses = [
ModelResponse(result=[AIMessage(content="")]),
ModelResponse(result=[AIMessage(content="still invalid")]),
]
request = MagicMock()
request.messages = [HumanMessage(content="Route this request.")]
corrected_request = object()
request.override.return_value = corrected_request
handler = AsyncMock(side_effect=responses)
middleware = StructuredResponseTextFallbackMiddleware(SourceRoutingPlan)

result = await middleware.awrap_model_call(request, handler)

assert result is responses[1]
assert result.structured_response is None
assert handler.await_args_list == [((request,),), ((corrected_request,),)]

@pytest.mark.asyncio
async def test_preserves_native_structured_response(self) -> None:
structured = SourceRoutingPlan.model_validate(self._routing())
response = ModelResponse(result=[AIMessage(content="")], structured_response=structured)
middleware = StructuredResponseTextFallbackMiddleware(SourceRoutingPlan)
handler = AsyncMock(return_value=response)

result = await middleware.awrap_model_call(None, handler)

assert result is response
handler.assert_awaited_once_with(None)

@pytest.mark.asyncio
async def test_does_not_intercept_tool_calls(self) -> None:
response = ModelResponse(
result=[
AIMessage(
content=json.dumps(self._routing()),
tool_calls=[{"name": "lookup_source_catalog", "args": {}, "id": "lookup-1"}],
)
]
)
middleware = StructuredResponseTextFallbackMiddleware(SourceRoutingPlan)

result = await middleware.awrap_model_call(None, AsyncMock(return_value=response))

assert result is response
assert result.structured_response is None


class TestSourceRoutingGuardMiddleware:
"""Tests for the orchestrator's required source-routing transition."""

Expand Down
34 changes: 34 additions & 0 deletions tests/aiq_agent/agents/deep_researcher/test_deepagents_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,40 @@
SYNTHESIS_SKILL_SOURCE = f"{BUILTIN_SKILL_SOURCE}synthesis/"


def test_frontier_profile_uses_validated_gpt_role_split() -> None:
"""Keep the shipped GPT profile aligned with the validated Sol/Luna topology."""
config = yaml.safe_load(Path("configs/config_frontier_models.yml").read_text(encoding="utf-8"))
llms = config["llms"]

assert llms["gpt_sol_agent_llm"] == {
"_type": "openai",
"model_name": "gpt-5.6-sol",
"api_key": "${OPENAI_API_KEY}",
"max_tokens": 16384,
"num_retries": 2,
"parallel_tool_calls": False,
}
assert llms["gpt_sol_writer_llm"] == {
**llms["gpt_sol_agent_llm"],
"max_tokens": 32768,
}
assert llms["gpt_luna_agent_llm"] == {
**llms["gpt_sol_agent_llm"],
"model_name": "gpt-5.6-luna",
}
deep_research = config["functions"]["deep_research_agent"]
assert {
role: deep_research[f"{role}_llm"]
for role in ("orchestrator", "source_router", "researcher", "planner", "writer")
} == {
"orchestrator": "gpt_sol_agent_llm",
"source_router": "gpt_luna_agent_llm",
"researcher": "gpt_luna_agent_llm",
"planner": "gpt_sol_agent_llm",
"writer": "gpt_sol_writer_llm",
}
Comment thread
AjayThorve marked this conversation as resolved.
Resolved


def test_openshell_workflow_only_diverges_for_sandbox_wiring() -> None:
"""Keep the OpenShell workflow aligned with the standard web config."""

Expand Down
Loading
Loading