开源通用 Agent - Suna 解析

@xml_schema(tag_name="tool_name", example="<tool_name>...</tool_name>")
def tool_method(self, param: str):
    # 工具实现
    return self.success_response(result)

# 文件: suna/backend/agent/prompt.py
SYSTEM_PROMPT = f"""
You are Suna.so, an autonomous AI Agent...

# 2.2 SYSTEM INFORMATION
- UTC DATE: {datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')}
- UTC TIME: {datetime.datetime.now(datetime.timezone.utc).strftime('%H:%M:%S')}

# 5.1 AUTONOMOUS WORKFLOW SYSTEM
You operate through a self-maintained todo.md file...

# 7.3 ATTACHMENT PROTOCOL
- CRITICAL: ALL VISUALIZATIONS MUST BE ATTACHED:
  * When using the 'ask' tool <ask attachments="file1, file2, file3"></ask>, ALWAYS attach ALL visualizations...
"""

def get_system_prompt():
    return SYSTEM_PROMPT

# 文件: suna/backend/agent/run.py
if "gemini-2.5-flash" in model_name.lower():
    # Gemini模型使用包含详细示例的Prompt
    system_message = { "role": "system", "content": get_gemini_system_prompt() }
elif "anthropic" not in model_name.lower():
    # 非Anthropic模型添加示例响应
    sample_response_path = os.path.join(os.path.dirname(__file__), 'sample_responses/1.txt')
    with open(sample_response_path, 'r') as file:
        sample_response = file.read()
    system_message = { "role": "system", "content": get_system_prompt() + "\n\n <sample_assistant_response>" + sample_response + "</sample_assistant_response>" }
else:
    # Anthropic模型使用基础Prompt
    system_message = { "role": "system", "content": get_system_prompt() }

# 文件: suna/backend/agentpress/tool.py
def xml_schema(tag_name: str, mappings: List[Dict[str, Any]] = None, example: str = None):
    """XML schema装饰器，定义工具的XML调用格式"""
    def decorator(func):
        # 创建XMLTagSchema实例
        xml_tag_def = XMLTagSchema(tag_name=tag_name, example=example)

        # 添加参数映射
        if mappings:
            for mapping_info in mappings:
                xml_tag_def.add_mapping(
                    param_name=mapping_info["param_name"],
                    node_type=mapping_info.get("node_type", "element"),
                    path=mapping_info.get("path", "."),
                    required=mapping_info.get("required", True)
                )

        # 将schema附加到函数
        tool_schema_entry = ToolSchema(schema_type=SchemaType.XML, xml_schema=xml_tag_def)
        return _add_schema(func, tool_schema_entry)
    return decorator

# 文件: suna/backend/run_agent_background.py
@dramatiq.actor
async def run_agent_background(
    agent_run_id: str,
    thread_id: str,
    instance_id: str,
    project_id: str,
    model_name: str,
    enable_thinking: Optional[bool],
    reasoning_effort: Optional[str],
    stream: bool,
    enable_context_manager: bool
):
    """在后台异步运行Agent，使用Redis进行状态管理"""
    await initialize()

    # 初始化Agent生成器
    agent_gen = run_agent(
        thread_id=thread_id,
        project_id=project_id,
        stream=stream,
        model_name=model_name,
        enable_thinking=enable_thinking,
        reasoning_effort=reasoning_effort,
        enable_context_manager=enable_context_manager
    )

    # 异步处理Agent响应
    async for response in agent_gen:
        # 将响应存储到Redis并发布通知
        await redis.rpush(response_list_key, response_json)
        await redis.publish(response_channel, "new")

# 文件: suna/backend/sandbox/sandbox.py
def create_sandbox(password: str, project_id: str = None):
    """创建新的沙盒环境"""
    params = CreateSandboxParams(
        image=Configuration.SANDBOX_IMAGE_NAME,
        public=True,
        env_vars={
            "CHROME_PERSISTENT_SESSION": "true",
            "RESOLUTION": "1024x768x24",
            "VNC_PASSWORD": password,
            "CHROME_DEBUGGING_PORT": "9222",
        },
        resources={
            "cpu": 2,
            "memory": 4,  # GB
            "disk": 5,    # GB
        }
    )

    sandbox = daytona.create(params)
    start_supervisord_session(sandbox)
    return sandbox