import base64
import json
import logging
import time
from rp.autonomous.detection import is_task_complete, get_completion_reason
from rp.autonomous.verification import TaskVerifier, create_task_verifier
from rp.config import STREAMING_ENABLED, VISIBLE_REASONING, VERIFICATION_REQUIRED
from rp.core.api import call_api
from rp.core.context import truncate_tool_result
from rp.core.cost_optimizer import CostOptimizer, create_cost_optimizer
from rp.core.debug import debug_trace
from rp.core.error_handler import ErrorHandler
from rp.core.reasoning import ReasoningEngine, ReasoningTrace
from rp.core.tool_selector import ToolSelector
from rp.tools.base import get_tools_definition
from rp.ui import Colors
from rp.ui.progress import ProgressIndicator
logger = logging.getLogger("rp")
def extract_reasoning_and_clean_content(content):
reasoning = None
lines = content.split("\n")
cleaned_lines = []
for line in lines:
if line.strip().startswith("REASONING:"):
reasoning = line.strip()[10:].strip()
else:
cleaned_lines.append(line)
cleaned_content = "\n".join(cleaned_lines)
cleaned_content = cleaned_content.replace("[TASK_COMPLETE]", "").strip()
return reasoning, cleaned_content
def sanitize_for_json(obj):
if isinstance(obj, bytes):
return base64.b64encode(obj).decode("utf-8")
elif isinstance(obj, dict):
return {k: sanitize_for_json(v) for k, v in obj.items()}
elif isinstance(obj, (list, tuple)):
return [sanitize_for_json(item) for item in obj]
else:
return obj
class AutonomousExecutor:
def __init__(self, assistant):
self.assistant = assistant
self.visible_reasoning = bool(assistant.verbose)
self.reasoning_engine = ReasoningEngine(visible=self.visible_reasoning)
self.tool_selector = ToolSelector()
self.error_handler = ErrorHandler()
self.cost_optimizer = create_cost_optimizer()
self.verifier = create_task_verifier(visible=self.visible_reasoning)
self.current_trace = None
self.tool_results = []
@debug_trace
def execute(self, task: str):
self.assistant.autonomous_mode = True
self.assistant.autonomous_iterations = 0
last_printed_result = None
logger.debug("=== AUTONOMOUS MODE START ===")
logger.debug(f"Task: {task}")
print(f"\n{Colors.BOLD}{Colors.CYAN}{'═' * 70}{Colors.RESET}")
print(f"{Colors.BOLD}Task:{Colors.RESET} {task[:100]}{'...' if len(task) > 100 else ''}")
print(f"{Colors.GRAY}Working autonomously. Press Ctrl+C to interrupt.{Colors.RESET}")
print(f"{Colors.BOLD}{Colors.CYAN}{'═' * 70}{Colors.RESET}\n")
self.current_trace = self.reasoning_engine.start_trace()
from rp.core.knowledge_context import inject_knowledge_context
self.current_trace.start_thinking()
intent = self.reasoning_engine.extract_intent(task)
self.current_trace.add_thinking(f"Task type: {intent['task_type']}, Complexity: {intent['complexity']}")
if intent['is_destructive']:
self.current_trace.add_thinking("Destructive operation detected - will proceed with caution")
if intent['requires_tools']:
selection = self.tool_selector.select(task, {'request': task})
self.current_trace.add_thinking(f"Tool strategy: {selection.reasoning}")
self.current_trace.add_thinking(f"Execution pattern: {selection.execution_pattern}")
self.current_trace.end_thinking()
optimizations = self.cost_optimizer.suggest_optimization(task, {
'message_count': len(self.assistant.messages),
'has_cache_prefix': hasattr(self.assistant, 'api_cache') and self.assistant.api_cache is not None
})
if optimizations and VISIBLE_REASONING:
for opt in optimizations:
if opt.estimated_savings > 0:
logger.debug(f"Optimization available: {opt.strategy.value} ({opt.estimated_savings:.0%} savings)")
self.assistant.messages.append({"role": "user", "content": f"{task}"})
if hasattr(self.assistant, "memory_manager"):
self.assistant.memory_manager.process_message(
task, role="user", extract_facts=True, update_graph=True
)
logger.debug("Extracted facts from user task and stored in memory")
inject_knowledge_context(self.assistant, self.assistant.messages[-1]["content"])
try:
while True:
self.assistant.autonomous_iterations += 1
iteration = self.assistant.autonomous_iterations
logger.debug(f"--- Autonomous iteration {iteration} ---")
logger.debug(f"Messages before context management: {len(self.assistant.messages)}")
if iteration > 1:
print(f"{Colors.GRAY}─── Iteration {iteration} ───{Colors.RESET}")
from rp.core.context import manage_context_window, refresh_system_message
self.assistant.messages = manage_context_window(self.assistant.messages, self.assistant.verbose)
logger.debug(f"Messages after context management: {len(self.assistant.messages)}")
with ProgressIndicator("Querying AI..."):
refresh_system_message(self.assistant.messages, self.assistant.args)
response = call_api(
self.assistant.messages,
self.assistant.model,
self.assistant.api_url,
self.assistant.api_key,
self.assistant.use_tools,
get_tools_definition(),
verbose=self.assistant.verbose,
)
if "usage" in response:
usage = response["usage"]
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
cached_tokens = usage.get("cached_tokens", 0)
cost_breakdown = self.cost_optimizer.calculate_cost(input_tokens, output_tokens, cached_tokens)
self.assistant.usage_tracker.track_request(self.assistant.model, input_tokens, output_tokens)
print(f"{Colors.YELLOW}💰 Cost: {self.cost_optimizer.format_cost(cost_breakdown.total_cost)} | "
f"Session: {self.cost_optimizer.format_cost(sum(c.total_cost for c in self.cost_optimizer.session_costs))}{Colors.RESET}")
if "error" in response:
logger.error(f"API error in autonomous mode: {response['error']}")
print(f"{Colors.RED}Error: {response['error']}{Colors.RESET}")
break
is_complete = is_task_complete(response, iteration)
if VERIFICATION_REQUIRED and is_complete:
verification = self.verifier.verify_task(
response, self.tool_results, task, iteration
)
if verification.needs_retry and iteration < 5:
is_complete = False
logger.info(f"Verification failed, retrying: {verification.retry_reason}")
logger.debug(f"Task completion check: {is_complete}")
if is_complete:
result = self._process_response(response)
if result != last_printed_result:
completion_reason = get_completion_reason(response, iteration)
if completion_reason and self.visible_reasoning:
print(f"{Colors.CYAN}[Completion: {completion_reason}]{Colors.RESET}")
print(f"\n{Colors.GREEN}r:{Colors.RESET} {result}\n")
last_printed_result = result
self._display_session_summary()
logger.debug(f"=== AUTONOMOUS MODE COMPLETE ===")
logger.debug(f"Total iterations: {iteration}")
logger.debug(f"Final message count: {len(self.assistant.messages)}")
break
result = self._process_response(response)
if result and result != last_printed_result:
print(f"\n{Colors.GREEN}r:{Colors.RESET} {result}\n")
last_printed_result = result
time.sleep(0.5)
except KeyboardInterrupt:
logger.debug("Autonomous mode interrupted by user")
print(f"\n{Colors.YELLOW}Autonomous mode interrupted by user{Colors.RESET}")
if self.assistant.messages and self.assistant.messages[-1]["role"] == "user":
self.assistant.messages.pop()
finally:
self.assistant.autonomous_mode = False
logger.debug("=== AUTONOMOUS MODE END ===")
def _process_response(self, response):
if "error" in response:
return f"Error: {response['error']}"
if "choices" not in response or not response["choices"]:
return "No response from API"
message = response["choices"][0]["message"]
self.assistant.messages.append(message)
if "tool_calls" in message and message["tool_calls"]:
self.current_trace.start_execution()
tool_results = []
for tool_call in message["tool_calls"]:
func_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
prevention = self.error_handler.prevent(func_name, arguments)
if prevention.blocked:
print(f"{Colors.RED}⚠ Blocked: {func_name} - {prevention.reason}{Colors.RESET}")
self.current_trace.add_tool_call(func_name, arguments, f"BLOCKED: {prevention.reason}", 0)
tool_results.append({
"tool_call_id": tool_call["id"],
"role": "tool",
"content": json.dumps({"status": "error", "error": f"Blocked: {prevention.reason}"})
})
continue
args_str = ", ".join([f"{k}={repr(v)[:50]}" for k, v in arguments.items()])
if len(args_str) > 80:
args_str = args_str[:77] + "..."
print(f"{Colors.BLUE}▶ {func_name}({args_str}){Colors.RESET}", end="", flush=True)
start_time = time.time()
result = execute_single_tool(self.assistant, func_name, arguments)
duration = time.time() - start_time
if isinstance(result, str):
try:
result = json.loads(result)
except json.JSONDecodeError as ex:
result = {"error": str(ex)}
is_error = isinstance(result, dict) and (result.get("status") == "error" or "error" in result)
if is_error:
print(f" {Colors.RED}✗ ({duration:.1f}s){Colors.RESET}")
error_msg = result.get("error", "Unknown error")[:100]
print(f" {Colors.RED}└─ {error_msg}{Colors.RESET}")
else:
print(f" {Colors.GREEN}✓ ({duration:.1f}s){Colors.RESET}")
errors = self.error_handler.detect(result, func_name)
if errors:
for error in errors:
recovery = self.error_handler.recover(
error, func_name, arguments,
lambda name, args: execute_single_tool(self.assistant, name, args)
)
self.error_handler.learn(error, recovery)
if recovery.success:
result = recovery.result
print(f" {Colors.GREEN}└─ Recovered: {recovery.message}{Colors.RESET}")
break
elif recovery.needs_human:
print(f" {Colors.YELLOW}└─ {recovery.error}{Colors.RESET}")
self.current_trace.add_tool_call(func_name, arguments, result, duration)
result = truncate_tool_result(result)
sanitized_result = sanitize_for_json(result)
tool_results.append({
"tool_call_id": tool_call["id"],
"role": "tool",
"content": json.dumps(sanitized_result),
})
self.tool_results.append({
'tool': func_name,
'status': result.get('status', 'unknown') if isinstance(result, dict) else 'success',
'result': result,
'duration': duration
})
self.current_trace.end_execution()
for result in tool_results:
self.assistant.messages.append(result)
with ProgressIndicator("Processing tool results..."):
from rp.core.context import refresh_system_message
refresh_system_message(self.assistant.messages, self.assistant.args)
follow_up = call_api(
self.assistant.messages,
self.assistant.model,
self.assistant.api_url,
self.assistant.api_key,
self.assistant.use_tools,
get_tools_definition(),
verbose=self.assistant.verbose,
)
if "usage" in follow_up:
usage = follow_up["usage"]
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
cached_tokens = usage.get("cached_tokens", 0)
self.cost_optimizer.calculate_cost(input_tokens, output_tokens, cached_tokens)
self.assistant.usage_tracker.track_request(self.assistant.model, input_tokens, output_tokens)
return self._process_response(follow_up)
content = message.get("content", "")
reasoning, cleaned_content = extract_reasoning_and_clean_content(content)
if reasoning and VISIBLE_REASONING:
print(f"{Colors.BLUE}💭 Reasoning: {reasoning}{Colors.RESET}")
from rp.ui import render_markdown
return render_markdown(cleaned_content, self.assistant.syntax_highlighting)
def _display_session_summary(self):
if not self.visible_reasoning:
return
summary = self.cost_optimizer.get_session_summary()
if summary.total_requests > 0:
print(f"\n{Colors.CYAN}━━━ Session Summary ━━━{Colors.RESET}")
print(f" Requests: {summary.total_requests}")
print(f" Total tokens: {summary.total_input_tokens + summary.total_output_tokens}")
print(f" Total cost: {self.cost_optimizer.format_cost(summary.total_cost)}")
if summary.total_savings > 0:
print(f" Savings: {self.cost_optimizer.format_cost(summary.total_savings)}")
error_stats = self.error_handler.get_statistics()
if error_stats.get('total_errors', 0) > 0:
print(f" Errors handled: {error_stats['total_errors']}")
trace_summary = self.current_trace.get_summary() if self.current_trace else {}
if trace_summary.get('execution_steps', 0) > 0:
print(f" Tool calls: {trace_summary['execution_steps']}")
print(f" Duration: {trace_summary['total_duration']:.1f}s")
print(f"{Colors.CYAN}━━━━━━━━━━━━━━━━━━━━━━━{Colors.RESET}\n")
def run_autonomous_mode(assistant, task):
executor = AutonomousExecutor(assistant)
executor.execute(task)
def process_response_autonomous(assistant, response):
if "error" in response:
return f"Error: {response['error']}"
if "choices" not in response or not response["choices"]:
return "No response from API"
message = response["choices"][0]["message"]
assistant.messages.append(message)
if "tool_calls" in message and message["tool_calls"]:
tool_results = []
for tool_call in message["tool_calls"]:
func_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
args_str = ", ".join([f"{k}={repr(v)[:50]}" for k, v in arguments.items()])
if len(args_str) > 80:
args_str = args_str[:77] + "..."
print(f"{Colors.BLUE}▶ {func_name}({args_str}){Colors.RESET}", end="", flush=True)
start_time = time.time()
result = execute_single_tool(assistant, func_name, arguments)
duration = time.time() - start_time
if isinstance(result, str):
try:
result = json.loads(result)
except json.JSONDecodeError as ex:
result = {"error": str(ex)}
is_error = isinstance(result, dict) and (result.get("status") == "error" or "error" in result)
if is_error:
print(f" {Colors.RED}✗ ({duration:.1f}s){Colors.RESET}")
error_msg = result.get("error", "Unknown error")[:100]
print(f" {Colors.RED}└─ {error_msg}{Colors.RESET}")
else:
print(f" {Colors.GREEN}✓ ({duration:.1f}s){Colors.RESET}")
result = truncate_tool_result(result)
sanitized_result = sanitize_for_json(result)
tool_results.append({
"tool_call_id": tool_call["id"],
"role": "tool",
"content": json.dumps(sanitized_result),
})
for result in tool_results:
assistant.messages.append(result)
with ProgressIndicator("Processing tool results..."):
from rp.core.context import refresh_system_message
refresh_system_message(assistant.messages, assistant.args)
follow_up = call_api(
assistant.messages,
assistant.model,
assistant.api_url,
assistant.api_key,
assistant.use_tools,
get_tools_definition(),
verbose=assistant.verbose,
)
if "usage" in follow_up:
usage = follow_up["usage"]
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
assistant.usage_tracker.track_request(assistant.model, input_tokens, output_tokens)
cost = assistant.usage_tracker._calculate_cost(assistant.model, input_tokens, output_tokens)
total_cost = assistant.usage_tracker.session_usage["estimated_cost"]
print(f"{Colors.YELLOW}Cost: ${cost:.4f} | Total: ${total_cost:.4f}{Colors.RESET}")
return process_response_autonomous(assistant, follow_up)
content = message.get("content", "")
reasoning, cleaned_content = extract_reasoning_and_clean_content(content)
if reasoning:
print(f"{Colors.BLUE}Reasoning: {reasoning}{Colors.RESET}")
from rp.ui import render_markdown
return render_markdown(cleaned_content, assistant.syntax_highlighting)
def execute_single_tool(assistant, func_name, arguments):
logger.debug(f"Executing tool in autonomous mode: {func_name}")
logger.debug(f"Tool arguments: {arguments}")
from rp.tools.base import get_func_map
func_map = get_func_map(db_conn=assistant.db_conn, python_globals=assistant.python_globals)
if func_name in func_map:
try:
result = func_map[func_name](**arguments)
logger.debug(f"Tool execution result: {str(result)[:200]}...")
return result
except Exception as e:
logger.error(f"Tool execution error: {str(e)}")
return {"status": "error", "error": str(e)}
else:
logger.error(f"Unknown function requested: {func_name}")
return {"status": "error", "error": f"Unknown function: {func_name}"}
Reference in New Issue View Git Blame Copy Permalink