Ling0925
/
suna
mirror of https://github.com/kortix-ai/suna.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

lint

This commit is contained in:
marko-kraemer 2025-07-07 00:32:25 +02:00
parent c1c3f6cce1
commit 587dfccd78
1 changed files with 114 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

199
backend/agent/run.py
View File

@ -319,7 +319,8 @@ async def run_agent(
data = latest_user_message.data[0]['content'] data = latest_user_message.data[0]['content']
if isinstance(data, str): if isinstance(data, str):
data = json.loads(data) data = json.loads(data)
trace.update(input=data['content']) if trace:
trace.update(input=data['content'])
while continue_execution and iteration_count < max_iterations: while continue_execution and iteration_count < max_iterations:
iteration_count += 1 iteration_count += 1
@ -329,7 +330,8 @@ async def run_agent(
can_run, message, subscription = await check_billing_status(client, account_id) can_run, message, subscription = await check_billing_status(client, account_id)
if not can_run: if not can_run:
error_msg = f"Billing limit reached: {message}" error_msg = f"Billing limit reached: {message}"
trace.event(name="billing_limit_reached", level="ERROR", status_message=(f"{error_msg}")) if trace:
trace.event(name="billing_limit_reached", level="ERROR", status_message=(f"{error_msg}"))
# Yield a special message to indicate billing limit reached # Yield a special message to indicate billing limit reached
yield { yield {
"type": "status", "type": "status",
@ -343,7 +345,8 @@ async def run_agent(
message_type = latest_message.data[0].get('type') message_type = latest_message.data[0].get('type')
if message_type == 'assistant': if message_type == 'assistant':
logger.info(f"Last message was from assistant, stopping execution") logger.info(f"Last message was from assistant, stopping execution")
trace.event(name="last_message_from_assistant", level="DEFAULT", status_message=(f"Last message was from assistant, stopping execution")) if trace:
trace.event(name="last_message_from_assistant", level="DEFAULT", status_message=(f"Last message was from assistant, stopping execution"))
continue_execution = False continue_execution = False
break break
@ -383,7 +386,8 @@ async def run_agent(
"format": "image/jpeg" "format": "image/jpeg"
} }
}) })
trace.event(name="screenshot_url_added_to_temporary_message", level="DEFAULT", status_message=(f"Screenshot URL added to temporary message.")) if trace:
trace.event(name="screenshot_url_added_to_temporary_message", level="DEFAULT", status_message=(f"Screenshot URL added to temporary message."))
elif screenshot_base64: elif screenshot_base64:
# Fallback to base64 if URL not available # Fallback to base64 if URL not available
temp_message_content_list.append({ temp_message_content_list.append({
@ -392,17 +396,21 @@ async def run_agent(
"url": f"data:image/jpeg;base64,{screenshot_base64}", "url": f"data:image/jpeg;base64,{screenshot_base64}",
} }
}) })
trace.event(name="screenshot_base64_added_to_temporary_message", level="WARNING", status_message=(f"Screenshot base64 added to temporary message. Prefer screenshot_url if available.")) if trace:
trace.event(name="screenshot_base64_added_to_temporary_message", level="WARNING", status_message=(f"Screenshot base64 added to temporary message. Prefer screenshot_url if available."))
else: else:
logger.warning("Browser state found but no screenshot data.") logger.warning("Browser state found but no screenshot data.")
trace.event(name="browser_state_found_but_no_screenshot_data", level="WARNING", status_message=(f"Browser state found but no screenshot data.")) if trace:
trace.event(name="browser_state_found_but_no_screenshot_data", level="WARNING", status_message=(f"Browser state found but no screenshot data."))
else: else:
logger.warning("Model is Gemini, Anthropic, or OpenAI, so not adding screenshot to temporary message.") logger.warning("Model is Gemini, Anthropic, or OpenAI, so not adding screenshot to temporary message.")
trace.event(name="model_is_gemini_anthropic_or_openai", level="WARNING", status_message=(f"Model is Gemini, Anthropic, or OpenAI, so not adding screenshot to temporary message.")) if trace:
trace.event(name="model_is_gemini_anthropic_or_openai", level="WARNING", status_message=(f"Model is Gemini, Anthropic, or OpenAI, so not adding screenshot to temporary message."))
except Exception as e: except Exception as e:
logger.error(f"Error parsing browser state: {e}") logger.error(f"Error parsing browser state: {e}")
trace.event(name="error_parsing_browser_state", level="ERROR", status_message=(f"{e}")) if trace:
trace.event(name="error_parsing_browser_state", level="ERROR", status_message=(f"{e}"))
# Get the latest image_context message (NEW) # Get the latest image_context message (NEW)
latest_image_context_msg = await client.table('messages').select('*').eq('thread_id', thread_id).eq('type', 'image_context').order('created_at', desc=True).limit(1).execute() latest_image_context_msg = await client.table('messages').select('*').eq('thread_id', thread_id).eq('type', 'image_context').order('created_at', desc=True).limit(1).execute()
@ -430,7 +438,8 @@ async def run_agent(
await client.table('messages').delete().eq('message_id', latest_image_context_msg.data[0]["message_id"]).execute() await client.table('messages').delete().eq('message_id', latest_image_context_msg.data[0]["message_id"]).execute()
except Exception as e: except Exception as e:
logger.error(f"Error parsing image context: {e}") logger.error(f"Error parsing image context: {e}")
trace.event(name="error_parsing_image_context", level="ERROR", status_message=(f"{e}")) if trace:
trace.event(name="error_parsing_image_context", level="ERROR", status_message=(f"{e}"))
# If we have any content, construct the temporary_message # If we have any content, construct the temporary_message
if temp_message_content_list: if temp_message_content_list:
@ -449,7 +458,7 @@ async def run_agent(
# Gemini 2.5 Pro has 64k max output tokens # Gemini 2.5 Pro has 64k max output tokens
max_tokens = 64000 max_tokens = 64000
generation = trace.generation(name="thread_manager.run_thread") generation = trace.generation(name="thread_manager.run_thread") if trace else None
try: try:
# Make the LLM call and process the response # Make the LLM call and process the response
response = await thread_manager.run_thread( response = await thread_manager.run_thread(
@ -480,7 +489,8 @@ async def run_agent(
if isinstance(response, dict) and "status" in response and response["status"] == "error": if isinstance(response, dict) and "status" in response and response["status"] == "error":
logger.error(f"Error response from run_thread: {response.get('message', 'Unknown error')}") logger.error(f"Error response from run_thread: {response.get('message', 'Unknown error')}")
trace.event(name="error_response_from_run_thread", level="ERROR", status_message=(f"{response.get('message', 'Unknown error')}")) if trace:
trace.event(name="error_response_from_run_thread", level="ERROR", status_message=(f"{response.get('message', 'Unknown error')}"))
yield response yield response
break break
@ -490,97 +500,114 @@ async def run_agent(
# Process the response # Process the response
error_detected = False error_detected = False
full_response = ""
try: try:
full_response = "" # Check if response is iterable (async generator) or a dict (error case)
async for chunk in response: if hasattr(response, '__aiter__') and not isinstance(response, dict):
# If we receive an error chunk, we should stop after this iteration async for chunk in response:
if isinstance(chunk, dict) and chunk.get('type') == 'status' and chunk.get('status') == 'error': # If we receive an error chunk, we should stop after this iteration
logger.error(f"Error chunk detected: {chunk.get('message', 'Unknown error')}") if isinstance(chunk, dict) and chunk.get('type') == 'status' and chunk.get('status') == 'error':
trace.event(name="error_chunk_detected", level="ERROR", status_message=(f"{chunk.get('message', 'Unknown error')}")) logger.error(f"Error chunk detected: {chunk.get('message', 'Unknown error')}")
error_detected = True if trace:
yield chunk # Forward the error chunk trace.event(name="error_chunk_detected", level="ERROR", status_message=(f"{chunk.get('message', 'Unknown error')}"))
continue # Continue processing other chunks but don't break yet error_detected = True
yield chunk # Forward the error chunk
# Check for termination signal in status messages continue # Continue processing other chunks but don't break yet
if chunk.get('type') == 'status':
try:
# Parse the metadata to check for termination signal
metadata = chunk.get('metadata', {})
if isinstance(metadata, str):
metadata = json.loads(metadata)
if metadata.get('agent_should_terminate'):
agent_should_terminate = True
logger.info("Agent termination signal detected in status message")
trace.event(name="agent_termination_signal_detected", level="DEFAULT", status_message="Agent termination signal detected in status message")
# Extract the tool name from the status content if available
content = chunk.get('content', {})
if isinstance(content, str):
content = json.loads(content)
if content.get('function_name'):
last_tool_call = content['function_name']
elif content.get('xml_tag_name'):
last_tool_call = content['xml_tag_name']
except Exception as e:
logger.debug(f"Error parsing status message for termination check: {e}")
# Check for XML versions like <ask>, <complete>, or <web-browser-takeover> in assistant content chunks # Check for termination signal in status messages
if chunk.get('type') == 'assistant' and 'content' in chunk: if chunk.get('type') == 'status':
try: try:
# The content field might be a JSON string or object # Parse the metadata to check for termination signal
content = chunk.get('content', '{}') metadata = chunk.get('metadata', {})
if isinstance(content, str): if isinstance(metadata, str):
assistant_content_json = json.loads(content) metadata = json.loads(metadata)
else:
assistant_content_json = content if metadata.get('agent_should_terminate'):
agent_should_terminate = True
logger.info("Agent termination signal detected in status message")
if trace:
trace.event(name="agent_termination_signal_detected", level="DEFAULT", status_message="Agent termination signal detected in status message")
# Extract the tool name from the status content if available
content = chunk.get('content', {})
if isinstance(content, str):
content = json.loads(content)
if content.get('function_name'):
last_tool_call = content['function_name']
elif content.get('xml_tag_name'):
last_tool_call = content['xml_tag_name']
except Exception as e:
logger.debug(f"Error parsing status message for termination check: {e}")
# Check for XML versions like <ask>, <complete>, or <web-browser-takeover> in assistant content chunks
if chunk.get('type') == 'assistant' and 'content' in chunk:
try:
# The content field might be a JSON string or object
content = chunk.get('content', '{}')
if isinstance(content, str):
assistant_content_json = json.loads(content)
else:
assistant_content_json = content
# The actual text content is nested within # The actual text content is nested within
assistant_text = assistant_content_json.get('content', '') assistant_text = assistant_content_json.get('content', '')
full_response += assistant_text full_response += assistant_text
if isinstance(assistant_text, str): if isinstance(assistant_text, str):
if '</ask>' in assistant_text or '</complete>' in assistant_text or '</web-browser-takeover>' in assistant_text: if '</ask>' in assistant_text or '</complete>' in assistant_text or '</web-browser-takeover>' in assistant_text:
if '</ask>' in assistant_text: if '</ask>' in assistant_text:
xml_tool = 'ask' xml_tool = 'ask'
elif '</complete>' in assistant_text: elif '</complete>' in assistant_text:
xml_tool = 'complete' xml_tool = 'complete'
elif '</web-browser-takeover>' in assistant_text: elif '</web-browser-takeover>' in assistant_text:
xml_tool = 'web-browser-takeover' xml_tool = 'web-browser-takeover'
last_tool_call = xml_tool last_tool_call = xml_tool
logger.info(f"Agent used XML tool: {xml_tool}") logger.info(f"Agent used XML tool: {xml_tool}")
trace.event(name="agent_used_xml_tool", level="DEFAULT", status_message=(f"Agent used XML tool: {xml_tool}")) if trace:
except json.JSONDecodeError: trace.event(name="agent_used_xml_tool", level="DEFAULT", status_message=(f"Agent used XML tool: {xml_tool}"))
# Handle cases where content might not be valid JSON except json.JSONDecodeError:
logger.warning(f"Warning: Could not parse assistant content JSON: {chunk.get('content')}") # Handle cases where content might not be valid JSON
trace.event(name="warning_could_not_parse_assistant_content_json", level="WARNING", status_message=(f"Warning: Could not parse assistant content JSON: {chunk.get('content')}")) logger.warning(f"Warning: Could not parse assistant content JSON: {chunk.get('content')}")
except Exception as e: if trace:
logger.error(f"Error processing assistant chunk: {e}") trace.event(name="warning_could_not_parse_assistant_content_json", level="WARNING", status_message=(f"Warning: Could not parse assistant content JSON: {chunk.get('content')}"))
trace.event(name="error_processing_assistant_chunk", level="ERROR", status_message=(f"Error processing assistant chunk: {e}")) except Exception as e:
logger.error(f"Error processing assistant chunk: {e}")
if trace:
trace.event(name="error_processing_assistant_chunk", level="ERROR", status_message=(f"Error processing assistant chunk: {e}"))
yield chunk yield chunk
else:
# Response is not iterable, likely an error dict
logger.error(f"Response is not iterable: {response}")
error_detected = True
# Check if we should stop based on the last tool call or error # Check if we should stop based on the last tool call or error
if error_detected: if error_detected:
logger.info(f"Stopping due to error detected in response") logger.info(f"Stopping due to error detected in response")
trace.event(name="stopping_due_to_error_detected_in_response", level="DEFAULT", status_message=(f"Stopping due to error detected in response")) if trace:
generation.end(output=full_response, status_message="error_detected", level="ERROR") trace.event(name="stopping_due_to_error_detected_in_response", level="DEFAULT", status_message=(f"Stopping due to error detected in response"))
if generation:
generation.end(output=full_response, status_message="error_detected", level="ERROR")
break break
if agent_should_terminate or last_tool_call in ['ask', 'complete', 'web-browser-takeover']: if agent_should_terminate or last_tool_call in ['ask', 'complete', 'web-browser-takeover']:
logger.info(f"Agent decided to stop with tool: {last_tool_call}") logger.info(f"Agent decided to stop with tool: {last_tool_call}")
trace.event(name="agent_decided_to_stop_with_tool", level="DEFAULT", status_message=(f"Agent decided to stop with tool: {last_tool_call}")) if trace:
generation.end(output=full_response, status_message="agent_stopped") trace.event(name="agent_decided_to_stop_with_tool", level="DEFAULT", status_message=(f"Agent decided to stop with tool: {last_tool_call}"))
if generation:
generation.end(output=full_response, status_message="agent_stopped")
continue_execution = False continue_execution = False
except Exception as e: except Exception as e:
# Just log the error and re-raise to stop all iterations # Just log the error and re-raise to stop all iterations
error_msg = f"Error during response streaming: {str(e)}" error_msg = f"Error during response streaming: {str(e)}"
logger.error(f"Error: {error_msg}") logger.error(f"Error: {error_msg}")
trace.event(name="error_during_response_streaming", level="ERROR", status_message=(f"Error during response streaming: {str(e)}")) if trace:
generation.end(output=full_response, status_message=error_msg, level="ERROR") trace.event(name="error_during_response_streaming", level="ERROR", status_message=(f"Error during response streaming: {str(e)}"))
if generation:
generation.end(output=full_response, status_message=error_msg, level="ERROR")
yield { yield {
"type": "status", "type": "status",
"status": "error", "status": "error",
@ -593,7 +620,8 @@ async def run_agent(
# Just log the error and re-raise to stop all iterations # Just log the error and re-raise to stop all iterations
error_msg = f"Error running thread: {str(e)}" error_msg = f"Error running thread: {str(e)}"
logger.error(f"Error: {error_msg}") logger.error(f"Error: {error_msg}")
trace.event(name="error_running_thread", level="ERROR", status_message=(f"Error running thread: {str(e)}")) if trace:
trace.event(name="error_running_thread", level="ERROR", status_message=(f"Error running thread: {str(e)}"))
yield { yield {
"type": "status", "type": "status",
"status": "error", "status": "error",
@ -601,6 +629,7 @@ async def run_agent(
} }
# Stop execution immediately on any error # Stop execution immediately on any error
break break
generation.end(output=full_response) if generation:
generation.end(output=full_response)
langfuse.flush() langfuse.flush()