diff --git a/sgr_deep_research/core/agents/__init__.py b/sgr_deep_research/core/agents/__init__.py
index 98e17243..261af3e7 100644
--- a/sgr_deep_research/core/agents/__init__.py
+++ b/sgr_deep_research/core/agents/__init__.py
@@ -1,5 +1,6 @@
"""Agents module for SGR Agent Core."""
+from sgr_deep_research.core.agents.prompt_based_sgr_agent import PromptBasedSGRAgent
from sgr_deep_research.core.agents.sgr_agent import SGRAgent
from sgr_deep_research.core.agents.sgr_auto_tool_calling_agent import SGRAutoToolCallingAgent
from sgr_deep_research.core.agents.sgr_so_tool_calling_agent import SGRSOToolCallingAgent
@@ -7,6 +8,7 @@
from sgr_deep_research.core.agents.tool_calling_agent import ToolCallingAgent
__all__ = [
+ "PromptBasedSGRAgent",
"SGRAgent",
"SGRAutoToolCallingAgent",
"SGRSOToolCallingAgent",
diff --git a/sgr_deep_research/core/agents/prompt_based_sgr_agent.py b/sgr_deep_research/core/agents/prompt_based_sgr_agent.py
new file mode 100644
index 00000000..85dce1d4
--- /dev/null
+++ b/sgr_deep_research/core/agents/prompt_based_sgr_agent.py
@@ -0,0 +1,251 @@
+"""Prompt-based SGR Agent that uses prompt instructions for schema definition.
+
+This agent implements the same reasoning flow as SGRAgent but uses prompt-based
+instructions to ensure output format instead of the response_format parameter.
+This makes it compatible with models that don't support structured outputs.
+"""
+
+import json
+import logging
+import re
+from typing import Type
+
+from openai import AsyncOpenAI
+
+from sgr_deep_research.core.agent_definition import ExecutionConfig, LLMConfig, PromptsConfig
+from sgr_deep_research.core.base_agent import BaseAgent
+from sgr_deep_research.core.tools import (
+ BaseTool,
+ ClarificationTool,
+ CreateReportTool,
+ FinalAnswerTool,
+ NextStepToolsBuilder,
+ NextStepToolStub,
+ WebSearchTool,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class PromptBasedSGRAgent(BaseAgent):
+ """Agent for deep research tasks using SGR framework with prompt-based schema definition.
+
+ This agent uses prompts to instruct the LLM on the expected output format,
+ making it compatible with models that don't support the response_format parameter.
+ """
+
+ name: str = "prompt_based_sgr_agent"
+
+ def __init__(
+ self,
+ task: str,
+ openai_client: AsyncOpenAI,
+ llm_config: LLMConfig,
+ prompts_config: PromptsConfig,
+ execution_config: ExecutionConfig,
+ toolkit: list[Type[BaseTool]] | None = None,
+ ):
+ super().__init__(
+ task=task,
+ openai_client=openai_client,
+ llm_config=llm_config,
+ prompts_config=prompts_config,
+ execution_config=execution_config,
+ toolkit=toolkit,
+ )
+ self.max_searches = execution_config.max_searches
+
+ def _get_tools_schema(self, tools: list[Type[BaseTool]]) -> str:
+ """Generate a schema description for available tools in a human-readable format."""
+ schema_parts = []
+ for tool in tools:
+ # Get the JSON schema for the tool
+ tool_schema = tool.model_json_schema()
+ properties = tool_schema.get("properties", {})
+ required = tool_schema.get("required", [])
+
+ # Format tool description
+ tool_desc = f"Tool: {tool.tool_name}\n"
+ tool_desc += f"Description: {tool.description}\n"
+ tool_desc += "Parameters:\n"
+
+ for prop_name, prop_info in properties.items():
+ if prop_name == "tool_name_discriminator":
+ continue
+ prop_type = prop_info.get("type", "string")
+ prop_desc = prop_info.get("description", "")
+ is_required = " (required)" if prop_name in required else " (optional)"
+ tool_desc += f" - {prop_name} ({prop_type}){is_required}: {prop_desc}\n"
+
+ schema_parts.append(tool_desc)
+
+ return "\n".join(schema_parts)
+
+ async def _prepare_context(self) -> list[dict]:
+ """Prepare conversation context with system prompt including tool schemas."""
+ from pathlib import Path
+
+ from sgr_deep_research.core.services.prompt_loader import PromptLoader
+
+ # Get the tools that would be available
+ tools = await self._get_available_tools()
+
+ # Generate schema for tools
+ tools_schema = self._get_tools_schema(tools)
+
+ # Get available tools description
+ available_tools_str_list = [
+ f"{i}. {tool.tool_name}: {tool.description}" for i, tool in enumerate(tools, start=1)
+ ]
+
+ # Load the prompt-based system prompt template
+ try:
+ # Find the prompts directory relative to this file
+ prompts_dir = Path(__file__).parent.parent / "prompts"
+ template_path = prompts_dir / "prompt_based_system_prompt.txt"
+
+ with open(template_path, "r") as f:
+ template = f.read()
+
+ system_prompt = template.format(
+ tools_schema=tools_schema,
+ available_tools="\n".join(available_tools_str_list),
+ )
+ except Exception as e:
+ logger.warning(f"Failed to load prompt-based system prompt: {e}, falling back to default")
+ system_prompt = PromptLoader.get_system_prompt(self.toolkit, self.prompts_config)
+
+ return [
+ {"role": "system", "content": system_prompt},
+ *self.conversation,
+ ]
+
+ async def _get_available_tools(self) -> list[Type[BaseTool]]:
+ """Get list of available tools based on current context."""
+ tools = set(self.toolkit)
+ if self._context.iteration >= self.max_iterations:
+ tools = {
+ CreateReportTool,
+ FinalAnswerTool,
+ }
+ if self._context.clarifications_used >= self.max_clarifications:
+ tools -= {
+ ClarificationTool,
+ }
+ if self._context.searches_used >= self.max_searches:
+ tools -= {
+ WebSearchTool,
+ }
+ return list(tools)
+
+ async def _prepare_tools(self) -> Type[NextStepToolStub]:
+ """Prepare tool classes with current context limits."""
+ tools = await self._get_available_tools()
+ return NextStepToolsBuilder.build_NextStepTools(tools)
+
+ def _parse_tool_call_from_response(self, response_text: str) -> dict:
+ """Parse tool call from LLM response text.
+
+ Expected format:
+
+ {"name": "tool_name", "arguments": {...}}
+
+ """
+ # Extract content between tags
+ pattern = r"\s*(\{.*?\})\s*"
+ matches = re.findall(pattern, response_text, re.DOTALL)
+
+ if not matches:
+ raise ValueError(f"No tool call found in response: {response_text[:200]}")
+
+ # Parse the JSON
+ try:
+ tool_call_json = json.loads(matches[0])
+ return tool_call_json
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Failed to parse tool call JSON: {e}, content: {matches[0][:200]}")
+
+ async def _reasoning_phase(self) -> NextStepToolStub:
+ """Execute reasoning phase using prompt-based tool calling."""
+ async with self.openai_client.chat.completions.stream(
+ model=self.llm_config.model,
+ messages=await self._prepare_context(),
+ max_tokens=self.llm_config.max_tokens,
+ temperature=self.llm_config.temperature,
+ ) as stream:
+ async for event in stream:
+ if event.type == "chunk":
+ self.streaming_generator.add_chunk(event.chunk)
+
+ # Get the complete response
+ completion = await stream.get_final_completion()
+ response_text = completion.choices[0].message.content
+
+ if not response_text:
+ raise ValueError("Empty response from LLM")
+
+ # Parse tool call from response
+ tool_call_data = self._parse_tool_call_from_response(response_text)
+
+ # Build the NextStepTool type dynamically
+ NextStepTool = await self._prepare_tools()
+
+ # Extract tool name and arguments
+ tool_name = tool_call_data.get("name")
+ tool_arguments = tool_call_data.get("arguments", {})
+
+ # Create the reasoning object with the tool function
+ # We need to reconstruct the full object that matches NextStepTool schema
+ reasoning_data = {
+ **tool_arguments,
+ "function": {
+ "tool_name_discriminator": tool_name,
+ **tool_arguments.get("function", {}),
+ },
+ }
+
+ try:
+ reasoning: NextStepToolStub = NextStepTool.model_validate(reasoning_data)
+ except Exception as e:
+ logger.error(f"Failed to validate reasoning: {e}, data: {reasoning_data}")
+ raise
+
+ self._log_reasoning(reasoning)
+ return reasoning
+
+ async def _select_action_phase(self, reasoning: NextStepToolStub) -> BaseTool:
+ """Select action tool from reasoning result."""
+ tool = reasoning.function
+ if not isinstance(tool, BaseTool):
+ raise ValueError("Selected tool is not a valid BaseTool instance")
+
+ self.conversation.append(
+ {
+ "role": "assistant",
+ "content": reasoning.remaining_steps[0] if reasoning.remaining_steps else "Completing",
+ "tool_calls": [
+ {
+ "type": "function",
+ "id": f"{self._context.iteration}-action",
+ "function": {
+ "name": tool.tool_name,
+ "arguments": tool.model_dump_json(),
+ },
+ }
+ ],
+ }
+ )
+ self.streaming_generator.add_tool_call(
+ f"{self._context.iteration}-action", tool.tool_name, tool.model_dump_json()
+ )
+ return tool
+
+ async def _action_phase(self, tool: BaseTool) -> str:
+ """Execute the selected tool."""
+ result = await tool(self._context)
+ self.conversation.append(
+ {"role": "tool", "content": result, "tool_call_id": f"{self._context.iteration}-action"}
+ )
+ self.streaming_generator.add_chunk_from_str(f"{result}\n")
+ self._log_tool_execution(tool, result)
+ return result
diff --git a/sgr_deep_research/core/prompts/prompt_based_system_prompt.txt b/sgr_deep_research/core/prompts/prompt_based_system_prompt.txt
new file mode 100644
index 00000000..cd909a00
--- /dev/null
+++ b/sgr_deep_research/core/prompts/prompt_based_system_prompt.txt
@@ -0,0 +1,65 @@
+
+You are an expert researcher with adaptive planning and schema-guided-reasoning capabilities. You get the research task and you neeed to do research and genrete answer
+
+
+
+PAY ATTENTION TO THE DATE INSIDE THE USER REQUEST
+DATE FORMAT: YYYY-MM-DD HH:MM:SS (ISO 8601)
+IMPORTANT: The date above is in YYYY-MM-DD format (Year-Month-Day). For example, 2025-10-03 means October 3rd, 2025, NOT March 10th.
+
+
+: Detect the language from user request and use this LANGUAGE for all responses, searches, and result finalanswertool
+LANGUAGE ADAPTATION: Always respond and create reports in the SAME LANGUAGE as the user's request.
+If user writes in Russian - respond in Russian, if in English - respond in English.
+:
+
+:
+1. Memorize plan you generated in first step and follow the task inside your plan.
+1. Adapt plan when new data contradicts initial assumptions
+2. Search queries in SAME LANGUAGE as user request
+3. Final Answer ENTIRELY in SAME LANGUAGE as user request
+
+
+
+ADAPTIVITY: Actively change plan when discovering new data.
+ANALYSIS EXTRACT DATA: Always analyze data that you took in extractpagecontenttool
+
+
+
+CRITICAL FOR FACTUAL ACCURACY:
+When answering questions about specific dates, numbers, versions, or names:
+1. EXACT VALUES: Extract the EXACT value from sources (day, month, year for dates; precise numbers for quantities)
+2. VERIFY YEAR: If question mentions a specific year (e.g., "in 2022"), verify extracted content is about that SAME year
+3. CROSS-VERIFICATION: When sources provide contradictory information, prefer:
+ - Official sources and primary documentation over secondary sources
+ - Search result snippets that DIRECTLY answer the question over extracted page content
+ - Multiple independent sources confirming the same fact
+4. DATE PRECISION: Pay special attention to exact dates - day matters (October 21 ≠ October 22)
+5. NUMBER PRECISION: For numbers/versions, exact match required (6.88b ≠ 6.88c, Episode 31 ≠ Episode 32)
+6. SNIPPET PRIORITY: If search snippet clearly states the answer, trust it unless extract proves it wrong
+7. TEMPORAL VALIDATION: When extracting page content, check if the page shows data for correct time period
+
+
+
+You may call one or more functions to assist with the user query.
+You are provided with function signatures within XML tags:
+
+{tools_schema}
+
+
+For each function call, return a json object with function name and arguments within XML tags:
+
+{{"name": , "arguments": }}
+
+
+IMPORTANT:
+- You MUST use one of the available tools for each step
+- The tool call must be valid JSON wrapped in tags
+- The "name" field must match one of the available tool names exactly
+- The "arguments" field must be a JSON object matching the tool's schema
+
+
+:
+Available tools:
+{available_tools}
+
diff --git a/tests/test_agent_factory.py b/tests/test_agent_factory.py
index f0677520..aa5bf400 100644
--- a/tests/test_agent_factory.py
+++ b/tests/test_agent_factory.py
@@ -16,6 +16,7 @@
)
from sgr_deep_research.core.agent_factory import AgentFactory
from sgr_deep_research.core.agents import (
+ PromptBasedSGRAgent,
SGRAgent,
SGRAutoToolCallingAgent,
SGRSOToolCallingAgent,
@@ -86,6 +87,7 @@ async def test_create_all_agent_types(self):
task = "Universal test task"
agent_classes = [
SGRAgent,
+ PromptBasedSGRAgent,
SGRToolCallingAgent,
SGRAutoToolCallingAgent,
SGRSOToolCallingAgent,
diff --git a/tests/test_prompt_based_sgr_agent.py b/tests/test_prompt_based_sgr_agent.py
new file mode 100644
index 00000000..353e3ee9
--- /dev/null
+++ b/tests/test_prompt_based_sgr_agent.py
@@ -0,0 +1,303 @@
+"""Tests for PromptBasedSGRAgent.
+
+This module contains tests for the PromptBasedSGRAgent class that uses
+prompt-based schema definition instead of response_format.
+"""
+
+from unittest.mock import Mock
+
+import pytest
+
+from sgr_deep_research.core.agent_definition import ExecutionConfig
+from sgr_deep_research.core.agents.prompt_based_sgr_agent import PromptBasedSGRAgent
+from sgr_deep_research.core.tools import (
+ FinalAnswerTool,
+ ReasoningTool,
+ WebSearchTool,
+)
+from tests.conftest import create_test_agent
+
+
+class TestPromptBasedSGRAgentInitialization:
+ """Tests for PromptBasedSGRAgent initialization."""
+
+ def test_initialization_basic(self):
+ """Test basic initialization."""
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test task",
+ execution_config=ExecutionConfig(max_iterations=20, max_clarifications=3, max_searches=10),
+ )
+
+ assert agent.task == "Test task"
+ assert agent.name == "prompt_based_sgr_agent"
+ assert agent.max_iterations == 20
+ assert agent.max_clarifications == 3
+ assert agent.max_searches == 10
+
+ def test_initialization_with_toolkit(self):
+ """Test initialization with custom toolkit."""
+ toolkit = [WebSearchTool, FinalAnswerTool]
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test",
+ toolkit=toolkit,
+ )
+
+ assert WebSearchTool in agent.toolkit
+ assert FinalAnswerTool in agent.toolkit
+
+
+class TestPromptBasedSGRAgentToolSchema:
+ """Tests for tool schema generation."""
+
+ def test_get_tools_schema(self):
+ """Test generating tool schema from tool classes."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+
+ tools = [ReasoningTool, WebSearchTool]
+ schema = agent._get_tools_schema(tools)
+
+ assert "Tool: reasoningtool" in schema
+ assert "Tool: websearchtool" in schema
+ assert "Description:" in schema
+ assert "Parameters:" in schema
+
+ def test_get_tools_schema_excludes_discriminator(self):
+ """Test that tool schema excludes tool_name_discriminator."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+
+ tools = [ReasoningTool]
+ schema = agent._get_tools_schema(tools)
+
+ assert "tool_name_discriminator" not in schema
+
+
+class TestPromptBasedSGRAgentToolSelection:
+ """Tests for tool selection and preparation."""
+
+ @pytest.mark.asyncio
+ async def test_get_available_tools_basic(self):
+ """Test getting available tools in basic state."""
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test",
+ toolkit=[WebSearchTool, FinalAnswerTool, ReasoningTool],
+ )
+
+ tools = await agent._get_available_tools()
+
+ assert WebSearchTool in tools
+ assert FinalAnswerTool in tools
+ assert ReasoningTool in tools
+
+ @pytest.mark.asyncio
+ async def test_get_available_tools_max_iterations_reached(self):
+ """Test that only completion tools are available at max iterations."""
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test",
+ toolkit=[WebSearchTool, FinalAnswerTool],
+ execution_config=ExecutionConfig(max_iterations=5),
+ )
+ agent._context.iteration = 5
+
+ tools = await agent._get_available_tools()
+
+ assert FinalAnswerTool in tools
+ assert WebSearchTool not in tools
+
+ @pytest.mark.asyncio
+ async def test_get_available_tools_max_searches_reached(self):
+ """Test that search tools are removed when max searches reached."""
+ from sgr_deep_research.core.tools import ClarificationTool
+
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test",
+ toolkit=[WebSearchTool, ClarificationTool, FinalAnswerTool],
+ execution_config=ExecutionConfig(max_searches=3),
+ )
+ agent._context.searches_used = 3
+
+ tools = await agent._get_available_tools()
+
+ assert WebSearchTool not in tools
+ assert ClarificationTool in tools
+ assert FinalAnswerTool in tools
+
+
+class TestPromptBasedSGRAgentToolCallParsing:
+ """Tests for parsing tool calls from LLM responses."""
+
+ def test_parse_tool_call_basic(self):
+ """Test parsing a basic tool call."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+
+ response = """
+ Let me search for that information.
+
+ {"name": "websearchtool", "arguments": {"query": "test query", "reasoning": "need info"}}
+
+ """
+
+ result = agent._parse_tool_call_from_response(response)
+
+ assert result["name"] == "websearchtool"
+ assert result["arguments"]["query"] == "test query"
+
+ def test_parse_tool_call_with_whitespace(self):
+ """Test parsing tool call with extra whitespace."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+
+ response = """
+
+ {
+ "name": "websearchtool",
+ "arguments": {
+ "query": "test query"
+ }
+ }
+
+ """
+
+ result = agent._parse_tool_call_from_response(response)
+
+ assert result["name"] == "websearchtool"
+
+ def test_parse_tool_call_no_tags(self):
+ """Test that parsing fails when no tool_call tags are found."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+
+ response = '{"name": "websearchtool", "arguments": {"query": "test"}}'
+
+ with pytest.raises(ValueError, match="No tool call found"):
+ agent._parse_tool_call_from_response(response)
+
+ def test_parse_tool_call_invalid_json(self):
+ """Test that parsing fails with invalid JSON."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+
+ response = """
+
+ {name: "websearchtool", arguments: {}}
+
+ """
+
+ with pytest.raises(ValueError, match="Failed to parse tool call JSON"):
+ agent._parse_tool_call_from_response(response)
+
+ def test_parse_tool_call_multiple_tags(self):
+ """Test parsing when multiple tool_call tags are present (uses first)."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+
+ response = """
+
+ {"name": "first_tool", "arguments": {}}
+
+ Some text
+
+ {"name": "second_tool", "arguments": {}}
+
+ """
+
+ result = agent._parse_tool_call_from_response(response)
+
+ # Should parse the first tool call
+ assert result["name"] == "first_tool"
+
+
+class TestPromptBasedSGRAgentContextPreparation:
+ """Tests for context preparation with prompt-based schema."""
+
+ @pytest.mark.asyncio
+ async def test_prepare_context_includes_tool_schema(self):
+ """Test that prepared context includes tool schemas."""
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test",
+ toolkit=[WebSearchTool, FinalAnswerTool],
+ )
+
+ context = await agent._prepare_context()
+
+ # Should have system message
+ assert len(context) >= 1
+ assert context[0]["role"] == "system"
+
+ # System message should contain tool information
+ system_content = context[0]["content"]
+ assert "tool_call" in system_content.lower() or "tools" in system_content.lower()
+
+ @pytest.mark.asyncio
+ async def test_prepare_context_with_conversation(self):
+ """Test context preparation with existing conversation."""
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test",
+ toolkit=[WebSearchTool],
+ )
+ agent.conversation = [
+ {"role": "user", "content": "test message"},
+ {"role": "assistant", "content": "test response"},
+ ]
+
+ context = await agent._prepare_context()
+
+ # Should include system + conversation
+ assert len(context) == 3
+ assert context[0]["role"] == "system"
+ assert context[1]["role"] == "user"
+ assert context[2]["role"] == "assistant"
+
+
+class TestPromptBasedSGRAgentIntegration:
+ """Integration tests for PromptBasedSGRAgent phases."""
+
+ @pytest.mark.asyncio
+ async def test_select_action_phase(self):
+ """Test select action phase with a valid reasoning result."""
+ agent = create_test_agent(
+ PromptBasedSGRAgent,
+ task="Test",
+ toolkit=[FinalAnswerTool],
+ )
+
+ # Create a mock reasoning result
+ reasoning = Mock(spec=ReasoningTool)
+ reasoning.function = FinalAnswerTool(
+ reasoning="Test complete",
+ completed_steps=["Step 1"],
+ answer="Final answer",
+ status="completed",
+ )
+ reasoning.remaining_steps = ["Complete task"]
+
+ tool = await agent._select_action_phase(reasoning)
+
+ assert isinstance(tool, FinalAnswerTool)
+ assert len(agent.conversation) == 1
+ assert agent.conversation[0]["role"] == "assistant"
+
+ @pytest.mark.asyncio
+ async def test_action_phase(self):
+ """Test action phase execution."""
+ agent = create_test_agent(PromptBasedSGRAgent, task="Test")
+ agent._context.iteration = 1
+
+ # Create a real tool instance for testing
+ from sgr_deep_research.core.models import AgentStatesEnum
+
+ tool = FinalAnswerTool(
+ reasoning="Test execution",
+ completed_steps=["Step 1"],
+ answer="Test answer",
+ status=AgentStatesEnum.COMPLETED,
+ )
+
+ result = await agent._action_phase(tool)
+
+ # Should be JSON string
+ assert "Test answer" in result
+ assert len(agent.conversation) == 1
+ assert agent.conversation[0]["role"] == "tool"