581 lines
24 KiB
Python
581 lines
24 KiB
Python
"""
|
|
ThinkDeep Workflow Tool - Extended Reasoning with Systematic Investigation
|
|
|
|
This tool provides step-by-step deep thinking capabilities using a systematic workflow approach.
|
|
It enables comprehensive analysis of complex problems with expert validation at completion.
|
|
|
|
Key Features:
|
|
- Systematic step-by-step thinking process
|
|
- Multi-step analysis with evidence gathering
|
|
- Confidence-based investigation flow
|
|
- Expert analysis integration with external models
|
|
- Support for focused analysis areas (architecture, performance, security, etc.)
|
|
- Confidence-based workflow optimization
|
|
"""
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any, Optional
|
|
|
|
from pydantic import Field
|
|
|
|
if TYPE_CHECKING:
|
|
from tools.models import ToolModelCategory
|
|
|
|
from config import TEMPERATURE_CREATIVE
|
|
from systemprompts import THINKDEEP_PROMPT
|
|
from tools.shared.base_models import WorkflowRequest
|
|
|
|
from .workflow.base import WorkflowTool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ThinkDeepWorkflowRequest(WorkflowRequest):
|
|
"""Request model for thinkdeep workflow tool with comprehensive investigation capabilities"""
|
|
|
|
# Core workflow parameters
|
|
step: str = Field(description="Current work step content and findings")
|
|
step_number: int = Field(description="Current step number (starts at 1)", ge=1)
|
|
total_steps: int = Field(description="Estimated total steps needed", ge=1)
|
|
next_step_required: bool = Field(description="Whether another step is needed")
|
|
findings: str = Field(
|
|
description="Discoveries: insights, connections, implications, evidence. "
|
|
"Document contradictions to earlier assumptions. Update past findings."
|
|
)
|
|
|
|
# Investigation tracking
|
|
files_checked: list[str] = Field(
|
|
default_factory=list,
|
|
description="All files examined (absolute paths). Include ruled-out files.",
|
|
)
|
|
relevant_files: list[str] = Field(
|
|
default_factory=list,
|
|
description="Files relevant to problem/goal (absolute paths). Include root cause, solution, key insights.",
|
|
)
|
|
relevant_context: list[str] = Field(
|
|
default_factory=list,
|
|
description="Key concepts/methods: 'concept_name' or 'ClassName.methodName'. Focus on core insights, decision points.",
|
|
)
|
|
hypothesis: Optional[str] = Field(
|
|
default=None,
|
|
description="Current theory based on evidence. Revise in later steps.",
|
|
)
|
|
|
|
# Analysis metadata
|
|
issues_found: list[dict] = Field(
|
|
default_factory=list,
|
|
description="Issues with dict: 'severity' (critical/high/medium/low), 'description'.",
|
|
)
|
|
confidence: str = Field(
|
|
default="low",
|
|
description="exploring/low/medium/high/very_high/almost_certain/certain. CRITICAL: 'certain' PREVENTS external validation.",
|
|
)
|
|
|
|
# Expert analysis configuration - keep these fields available for configuring the final assistant model
|
|
# in expert analysis (commented out exclude=True)
|
|
temperature: Optional[float] = Field(
|
|
default=None,
|
|
description="Creative thinking temp (0-1, default 0.7)",
|
|
ge=0.0,
|
|
le=1.0,
|
|
)
|
|
thinking_mode: Optional[str] = Field(
|
|
default=None,
|
|
description="Depth: minimal/low/medium/high/max. Default 'high'.",
|
|
)
|
|
# Context files and investigation scope
|
|
problem_context: Optional[str] = Field(
|
|
default=None,
|
|
description="Additional context about problem/goal. Be expressive.",
|
|
)
|
|
focus_areas: Optional[list[str]] = Field(
|
|
default=None,
|
|
description="Focus aspects (architecture, performance, security, etc.)",
|
|
)
|
|
|
|
|
|
class ThinkDeepTool(WorkflowTool):
|
|
"""
|
|
ThinkDeep Workflow Tool - Systematic Deep Thinking Analysis
|
|
|
|
Provides comprehensive step-by-step thinking capabilities with expert validation.
|
|
Uses workflow architecture for systematic investigation and analysis.
|
|
"""
|
|
|
|
name = "thinkdeep"
|
|
description = (
|
|
"Performs multi-stage investigation and reasoning for complex problem analysis. "
|
|
"Use for architecture decisions, complex bugs, performance challenges, and security analysis. "
|
|
"Provides systematic hypothesis testing, evidence-based investigation, and expert validation."
|
|
)
|
|
|
|
def __init__(self):
|
|
"""Initialize the ThinkDeep workflow tool"""
|
|
super().__init__()
|
|
# Storage for request parameters to use in expert analysis
|
|
self.stored_request_params = {}
|
|
|
|
def get_name(self) -> str:
|
|
"""Return the tool name"""
|
|
return self.name
|
|
|
|
def get_description(self) -> str:
|
|
"""Return the tool description"""
|
|
return self.description
|
|
|
|
def get_model_category(self) -> "ToolModelCategory":
|
|
"""Return the model category for this tool"""
|
|
from tools.models import ToolModelCategory
|
|
|
|
return ToolModelCategory.EXTENDED_REASONING
|
|
|
|
def get_workflow_request_model(self):
|
|
"""Return the workflow request model for this tool"""
|
|
return ThinkDeepWorkflowRequest
|
|
|
|
def get_input_schema(self) -> dict[str, Any]:
|
|
"""Generate input schema using WorkflowSchemaBuilder with thinkdeep-specific overrides."""
|
|
from .workflow.schema_builders import WorkflowSchemaBuilder
|
|
|
|
# ThinkDeep workflow-specific field overrides
|
|
thinkdeep_field_overrides = {
|
|
"problem_context": {
|
|
"type": "string",
|
|
"description": "Additional context about problem/goal. Be expressive.",
|
|
},
|
|
"focus_areas": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"description": "Focus aspects (architecture, performance, security, etc.)",
|
|
},
|
|
}
|
|
|
|
# Use WorkflowSchemaBuilder with thinkdeep-specific tool fields
|
|
return WorkflowSchemaBuilder.build_schema(
|
|
tool_specific_fields=thinkdeep_field_overrides,
|
|
model_field_schema=self.get_model_field_schema(),
|
|
auto_mode=self.is_effective_auto_mode(),
|
|
tool_name=self.get_name(),
|
|
)
|
|
|
|
def get_system_prompt(self) -> str:
|
|
"""Return the system prompt for this workflow tool"""
|
|
return THINKDEEP_PROMPT
|
|
|
|
def get_default_temperature(self) -> float:
|
|
"""Return default temperature for deep thinking"""
|
|
return TEMPERATURE_CREATIVE
|
|
|
|
def get_default_thinking_mode(self) -> str:
|
|
"""Return default thinking mode for thinkdeep"""
|
|
from config import DEFAULT_THINKING_MODE_THINKDEEP
|
|
|
|
return DEFAULT_THINKING_MODE_THINKDEEP
|
|
|
|
def customize_workflow_response(self, response_data: dict, request, **kwargs) -> dict:
|
|
"""
|
|
Customize the workflow response for thinkdeep-specific needs
|
|
"""
|
|
# Store request parameters for later use in expert analysis
|
|
self.stored_request_params = {}
|
|
try:
|
|
self.stored_request_params["temperature"] = request.temperature
|
|
except AttributeError:
|
|
self.stored_request_params["temperature"] = None
|
|
|
|
try:
|
|
self.stored_request_params["thinking_mode"] = request.thinking_mode
|
|
except AttributeError:
|
|
self.stored_request_params["thinking_mode"] = None
|
|
|
|
# Add thinking-specific context to response
|
|
response_data.update(
|
|
{
|
|
"thinking_status": {
|
|
"current_step": request.step_number,
|
|
"total_steps": request.total_steps,
|
|
"files_checked": len(request.files_checked),
|
|
"relevant_files": len(request.relevant_files),
|
|
"thinking_confidence": request.confidence,
|
|
"analysis_focus": request.focus_areas or ["general"],
|
|
}
|
|
}
|
|
)
|
|
|
|
# Add thinking_complete field for final steps (test expects this)
|
|
if not request.next_step_required:
|
|
response_data["thinking_complete"] = True
|
|
|
|
# Add complete_thinking summary (test expects this)
|
|
response_data["complete_thinking"] = {
|
|
"steps_completed": len(self.work_history),
|
|
"final_confidence": request.confidence,
|
|
"relevant_context": list(self.consolidated_findings.relevant_context),
|
|
"key_findings": self.consolidated_findings.findings,
|
|
"issues_identified": self.consolidated_findings.issues_found,
|
|
"files_analyzed": list(self.consolidated_findings.relevant_files),
|
|
}
|
|
|
|
# Add thinking-specific completion message based on confidence
|
|
if request.confidence == "certain":
|
|
response_data["completion_message"] = (
|
|
"Deep thinking analysis is complete with high certainty. "
|
|
"All aspects have been thoroughly considered and conclusions are definitive."
|
|
)
|
|
elif not request.next_step_required:
|
|
response_data["completion_message"] = (
|
|
"Deep thinking analysis phase complete. Expert validation will provide additional insights and recommendations."
|
|
)
|
|
|
|
return response_data
|
|
|
|
def should_skip_expert_analysis(self, request, consolidated_findings) -> bool:
|
|
"""
|
|
ThinkDeep tool skips expert analysis when the CLI agent has "certain" confidence.
|
|
"""
|
|
return request.confidence == "certain" and not request.next_step_required
|
|
|
|
def get_completion_status(self) -> str:
|
|
"""ThinkDeep tools use thinking-specific status."""
|
|
return "deep_thinking_complete_ready_for_implementation"
|
|
|
|
def get_completion_data_key(self) -> str:
|
|
"""ThinkDeep uses 'complete_thinking' key."""
|
|
return "complete_thinking"
|
|
|
|
def get_final_analysis_from_request(self, request):
|
|
"""ThinkDeep tools use 'findings' field."""
|
|
return request.findings
|
|
|
|
def get_skip_expert_analysis_status(self) -> str:
|
|
"""Status when skipping expert analysis for certain confidence."""
|
|
return "skipped_due_to_certain_thinking_confidence"
|
|
|
|
def get_skip_reason(self) -> str:
|
|
"""Reason for skipping expert analysis."""
|
|
return "Expressed 'certain' confidence in the deep thinking analysis - no additional validation needed"
|
|
|
|
def get_completion_message(self) -> str:
|
|
"""Message for completion without expert analysis."""
|
|
return "Deep thinking analysis complete with certain confidence. Proceed with implementation based on the analysis."
|
|
|
|
def customize_expert_analysis_prompt(self, base_prompt: str, request, file_content: str = "") -> str:
|
|
"""
|
|
Customize the expert analysis prompt for deep thinking validation
|
|
"""
|
|
thinking_context = f"""
|
|
DEEP THINKING ANALYSIS VALIDATION
|
|
|
|
You are reviewing a comprehensive deep thinking analysis completed through systematic investigation.
|
|
Your role is to validate the thinking process, identify any gaps, challenge assumptions, and provide
|
|
additional insights or alternative perspectives.
|
|
|
|
ANALYSIS SCOPE:
|
|
- Problem Context: {self._get_problem_context(request)}
|
|
- Focus Areas: {', '.join(self._get_focus_areas(request))}
|
|
- Investigation Confidence: {request.confidence}
|
|
- Steps Completed: {request.step_number} of {request.total_steps}
|
|
|
|
THINKING SUMMARY:
|
|
{request.findings}
|
|
|
|
KEY INSIGHTS AND CONTEXT:
|
|
{', '.join(request.relevant_context) if request.relevant_context else 'No specific context identified'}
|
|
|
|
VALIDATION OBJECTIVES:
|
|
1. Assess the depth and quality of the thinking process
|
|
2. Identify any logical gaps, missing considerations, or flawed assumptions
|
|
3. Suggest alternative approaches or perspectives not considered
|
|
4. Validate the conclusions and recommendations
|
|
5. Provide actionable next steps for implementation
|
|
|
|
Be thorough but constructive in your analysis. Challenge the thinking where appropriate,
|
|
but also acknowledge strong insights and valid conclusions.
|
|
"""
|
|
|
|
if file_content:
|
|
thinking_context += f"\n\nFILE CONTEXT:\n{file_content}"
|
|
|
|
return f"{thinking_context}\n\n{base_prompt}"
|
|
|
|
def get_expert_analysis_instructions(self) -> str:
|
|
"""
|
|
Return instructions for expert analysis specific to deep thinking validation
|
|
"""
|
|
return (
|
|
"DEEP THINKING ANALYSIS IS COMPLETE. You MUST now summarize and present ALL thinking insights, "
|
|
"alternative approaches considered, risks and trade-offs identified, and final recommendations. "
|
|
"Clearly prioritize the top solutions or next steps that emerged from the analysis. "
|
|
"Provide concrete, actionable guidance based on the deep thinking—make it easy for the user to "
|
|
"understand exactly what to do next and how to implement the best solution."
|
|
)
|
|
|
|
# Override hook methods to use stored request parameters for expert analysis
|
|
|
|
def get_request_temperature(self, request) -> float:
|
|
"""Use stored temperature from initial request."""
|
|
try:
|
|
stored_params = self.stored_request_params
|
|
if stored_params and stored_params.get("temperature") is not None:
|
|
return stored_params["temperature"]
|
|
except AttributeError:
|
|
pass
|
|
return super().get_request_temperature(request)
|
|
|
|
def get_request_thinking_mode(self, request) -> str:
|
|
"""Use stored thinking mode from initial request."""
|
|
try:
|
|
stored_params = self.stored_request_params
|
|
if stored_params and stored_params.get("thinking_mode") is not None:
|
|
return stored_params["thinking_mode"]
|
|
except AttributeError:
|
|
pass
|
|
return super().get_request_thinking_mode(request)
|
|
|
|
def _get_problem_context(self, request) -> str:
|
|
"""Get problem context from request. Override for custom context handling."""
|
|
try:
|
|
return request.problem_context or "General analysis"
|
|
except AttributeError:
|
|
return "General analysis"
|
|
|
|
def _get_focus_areas(self, request) -> list[str]:
|
|
"""Get focus areas from request. Override for custom focus area handling."""
|
|
try:
|
|
return request.focus_areas or ["comprehensive analysis"]
|
|
except AttributeError:
|
|
return ["comprehensive analysis"]
|
|
|
|
def get_required_actions(
|
|
self, step_number: int, confidence: str, findings: str, total_steps: int, request=None
|
|
) -> list[str]:
|
|
"""
|
|
Return required actions for the current thinking step.
|
|
"""
|
|
actions = []
|
|
|
|
if step_number == 1:
|
|
actions.extend(
|
|
[
|
|
"Begin systematic thinking analysis",
|
|
"Identify key aspects and assumptions to explore",
|
|
"Establish initial investigation approach",
|
|
]
|
|
)
|
|
elif confidence == "low":
|
|
actions.extend(
|
|
[
|
|
"Continue gathering evidence and insights",
|
|
"Test initial hypotheses",
|
|
"Explore alternative perspectives",
|
|
]
|
|
)
|
|
elif confidence == "medium":
|
|
actions.extend(
|
|
[
|
|
"Deepen analysis of promising approaches",
|
|
"Validate key assumptions",
|
|
"Consider implementation challenges",
|
|
]
|
|
)
|
|
elif confidence == "high":
|
|
actions.extend(
|
|
[
|
|
"Refine and validate key findings",
|
|
"Explore edge cases and limitations",
|
|
"Document assumptions and trade-offs",
|
|
]
|
|
)
|
|
elif confidence == "very_high":
|
|
actions.extend(
|
|
[
|
|
"Synthesize findings into cohesive recommendations",
|
|
"Validate conclusions against all evidence",
|
|
"Prepare comprehensive implementation guidance",
|
|
]
|
|
)
|
|
elif confidence == "almost_certain":
|
|
actions.extend(
|
|
[
|
|
"Finalize recommendations with high confidence",
|
|
"Document any remaining minor uncertainties",
|
|
"Prepare for expert analysis or implementation",
|
|
]
|
|
)
|
|
else: # certain
|
|
actions.append("Analysis complete - ready for implementation")
|
|
|
|
return actions
|
|
|
|
def should_call_expert_analysis(self, consolidated_findings, request=None) -> bool:
|
|
"""
|
|
Determine if expert analysis should be called based on confidence and completion.
|
|
"""
|
|
if request:
|
|
try:
|
|
# Don't call expert analysis if confidence is "certain"
|
|
if request.confidence == "certain":
|
|
return False
|
|
except AttributeError:
|
|
pass
|
|
|
|
# Call expert analysis if investigation is complete (when next_step_required is False)
|
|
if request:
|
|
try:
|
|
return not request.next_step_required
|
|
except AttributeError:
|
|
pass
|
|
|
|
# Fallback: call expert analysis if we have meaningful findings
|
|
return (
|
|
len(consolidated_findings.relevant_files) > 0
|
|
or len(consolidated_findings.findings) >= 2
|
|
or len(consolidated_findings.issues_found) > 0
|
|
)
|
|
|
|
def prepare_expert_analysis_context(self, consolidated_findings) -> str:
|
|
"""
|
|
Prepare context for expert analysis specific to deep thinking.
|
|
"""
|
|
context_parts = []
|
|
|
|
context_parts.append("DEEP THINKING ANALYSIS SUMMARY:")
|
|
context_parts.append(f"Steps completed: {len(consolidated_findings.findings)}")
|
|
context_parts.append(f"Final confidence: {consolidated_findings.confidence}")
|
|
|
|
if consolidated_findings.findings:
|
|
context_parts.append("\nKEY FINDINGS:")
|
|
for i, finding in enumerate(consolidated_findings.findings, 1):
|
|
context_parts.append(f"{i}. {finding}")
|
|
|
|
if consolidated_findings.relevant_context:
|
|
context_parts.append(f"\nRELEVANT CONTEXT:\n{', '.join(consolidated_findings.relevant_context)}")
|
|
|
|
# Get hypothesis from latest hypotheses entry if available
|
|
if consolidated_findings.hypotheses:
|
|
latest_hypothesis = consolidated_findings.hypotheses[-1].get("hypothesis", "")
|
|
if latest_hypothesis:
|
|
context_parts.append(f"\nFINAL HYPOTHESIS:\n{latest_hypothesis}")
|
|
|
|
if consolidated_findings.issues_found:
|
|
context_parts.append(f"\nISSUES IDENTIFIED: {len(consolidated_findings.issues_found)} issues")
|
|
for issue in consolidated_findings.issues_found:
|
|
context_parts.append(
|
|
f"- {issue.get('severity', 'unknown')}: {issue.get('description', 'No description')}"
|
|
)
|
|
|
|
return "\n".join(context_parts)
|
|
|
|
def get_step_guidance_message(self, request) -> str:
|
|
"""
|
|
Generate guidance for the next step in thinking analysis
|
|
"""
|
|
if request.next_step_required:
|
|
next_step_number = request.step_number + 1
|
|
|
|
if request.confidence == "certain":
|
|
guidance = (
|
|
f"Your thinking analysis confidence is CERTAIN. Consider if you truly need step {next_step_number} "
|
|
f"or if you should complete the analysis now with expert validation."
|
|
)
|
|
elif request.confidence == "almost_certain":
|
|
guidance = (
|
|
f"Your thinking analysis confidence is ALMOST_CERTAIN. For step {next_step_number}, consider: "
|
|
f"finalizing recommendations, documenting any minor uncertainties, or preparing for implementation."
|
|
)
|
|
elif request.confidence == "very_high":
|
|
guidance = (
|
|
f"Your thinking analysis confidence is VERY_HIGH. For step {next_step_number}, consider: "
|
|
f"synthesis of all findings, comprehensive validation, or creating implementation roadmap."
|
|
)
|
|
elif request.confidence == "high":
|
|
guidance = (
|
|
f"Your thinking analysis confidence is HIGH. For step {next_step_number}, consider: "
|
|
f"exploring edge cases, documenting trade-offs, or stress-testing key assumptions."
|
|
)
|
|
elif request.confidence == "medium":
|
|
guidance = (
|
|
f"Your thinking analysis confidence is MEDIUM. For step {next_step_number}, focus on: "
|
|
f"deepening insights, exploring alternative approaches, or gathering additional evidence."
|
|
)
|
|
else: # low or exploring
|
|
guidance = (
|
|
f"Your thinking analysis confidence is {request.confidence.upper()}. For step {next_step_number}, "
|
|
f"continue investigating: gather more evidence, test hypotheses, or explore different angles."
|
|
)
|
|
|
|
# Add specific thinking guidance based on progress
|
|
if request.step_number == 1:
|
|
guidance += (
|
|
" Consider: What are the key assumptions? What evidence supports or contradicts initial theories? "
|
|
"What alternative approaches exist?"
|
|
)
|
|
elif request.step_number >= request.total_steps // 2:
|
|
guidance += (
|
|
" Consider: Synthesis of findings, validation of conclusions, identification of implementation "
|
|
"challenges, and preparation for expert analysis."
|
|
)
|
|
|
|
return guidance
|
|
else:
|
|
return "Thinking analysis is ready for expert validation and final recommendations."
|
|
|
|
def format_final_response(self, assistant_response: str, request, **kwargs) -> dict:
|
|
"""
|
|
Format the final response from the assistant for thinking analysis
|
|
"""
|
|
response_data = {
|
|
"thinking_analysis": assistant_response,
|
|
"analysis_metadata": {
|
|
"total_steps_completed": request.step_number,
|
|
"final_confidence": request.confidence,
|
|
"files_analyzed": len(request.relevant_files),
|
|
"key_insights": len(request.relevant_context),
|
|
"issues_identified": len(request.issues_found),
|
|
},
|
|
}
|
|
|
|
# Add completion status
|
|
if request.confidence == "certain":
|
|
response_data["completion_status"] = "analysis_complete_with_certainty"
|
|
else:
|
|
response_data["completion_status"] = "analysis_complete_pending_validation"
|
|
|
|
return response_data
|
|
|
|
def format_step_response(
|
|
self,
|
|
assistant_response: str,
|
|
request,
|
|
status: str = "pause_for_thinkdeep",
|
|
continuation_id: Optional[str] = None,
|
|
**kwargs,
|
|
) -> dict:
|
|
"""
|
|
Format intermediate step responses for thinking workflow
|
|
"""
|
|
response_data = super().format_step_response(assistant_response, request, status, continuation_id, **kwargs)
|
|
|
|
# Add thinking-specific step guidance
|
|
step_guidance = self.get_step_guidance_message(request)
|
|
response_data["thinking_guidance"] = step_guidance
|
|
|
|
# Add analysis progress indicators
|
|
response_data["analysis_progress"] = {
|
|
"step_completed": request.step_number,
|
|
"remaining_steps": max(0, request.total_steps - request.step_number),
|
|
"confidence_trend": request.confidence,
|
|
"investigation_depth": "expanding" if request.next_step_required else "finalizing",
|
|
}
|
|
|
|
return response_data
|
|
|
|
# Required abstract methods from BaseTool
|
|
def get_request_model(self):
|
|
"""Return the thinkdeep workflow-specific request model."""
|
|
return ThinkDeepWorkflowRequest
|
|
|
|
async def prepare_prompt(self, request) -> str:
|
|
"""Not used - workflow tools use execute_workflow()."""
|
|
return "" # Workflow tools use execute_workflow() directly
|