AutoGPT框架完整使用手册：从入门到精通的开发指南

# 克隆项目
git clone https://github.com/Significant-Gravitas/AutoGPT.git
cd AutoGPT/autogpt_platform

# 安装依赖
cd backend && poetry install
cd ../frontend && pnpm install

# 启动服务
docker-compose up -d  # 启动数据库和Redis
cd backend && poetry run python -m backend.app  # 启动后端
cd ../frontend && pnpm dev  # 启动前端

graph TB
    subgraph "AutoGPT核心概念"
        Graph["Graph 工作流图"]
        Node["Node 执行节点"]
        Block["Block 功能块"]
        Execution["Execution 执行实例"]
        
        Graph --> Node
        Node --> Block
        Graph --> Execution
        Execution --> Node
    end
    
    subgraph "数据流向"
        Input["输入数据"]
        Output["输出结果"]
        
        Input --> Graph
        Graph --> Output
    end

// 前端创建图的基本流程
interface CreateGraphRequest {
  name: string;
  description?: string;
  nodes: GraphNode[];
  links: GraphLink[];
}

// 示例：创建一个简单的文本处理工作流
const simpleGraph: CreateGraphRequest = {
  name: "文本摘要工作流",
  description: "使用LLM对输入文本进行摘要",
  nodes: [
    {
      id: "input_node",
      block_id: "text_input_block",
      input_default: {},
      metadata: { position: { x: 100, y: 100 } }
    },
    {
      id: "llm_node", 
      block_id: "llm_call_block",
      input_default: {
        model: "gpt-4",
        prompt: "请对以下文本进行摘要：{text}"
      },
      metadata: { position: { x: 300, y: 100 } }
    }
  ],
  links: [
    {
      source_id: "input_node",
      sink_id: "llm_node",
      source_name: "text",
      sink_name: "text"
    }
  ]
};

import httpx
import asyncio

class AutoGPTClient:
    """AutoGPT平台客户端封装类
    
    提供便捷的API调用接口，包含认证、错误处理等功能
    """
    
    def __init__(self, base_url: str, api_key: str):
        """初始化客户端
        
        Args:
            base_url: API服务器地址，如 "https://api.autogpt.com"
            api_key: 用户API密钥，用于身份认证
        """
        self.base_url = base_url.rstrip('/')
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.client = httpx.AsyncClient(headers=self.headers)
    
    async def create_graph(self, graph_data: dict) -> dict:
        """创建新的工作流图
        
        Args:
            graph_data: 图定义数据，包含节点、连接等信息
            
        Returns:
            创建成功的图信息，包含图ID和版本号
            
        Raises:
            httpx.HTTPStatusError: API调用失败时抛出
        """
        response = await self.client.post(
            f"{self.base_url}/api/graphs",
            json=graph_data
        )
        response.raise_for_status()
        return response.json()
    
    async def execute_graph(self, graph_id: str, inputs: dict, 
                          graph_version: int = None) -> dict:
        """执行工作流图
        
        Args:
            graph_id: 图的唯一标识符
            inputs: 执行输入数据，键值对格式
            graph_version: 图版本号，默认使用最新版本
            
        Returns:
            执行结果信息，包含执行ID和状态
        """
        url = f"{self.base_url}/api/graphs/{graph_id}/execute"
        if graph_version:
            url += f"/{graph_version}"
            
        response = await self.client.post(url, json={
            "inputs": inputs,
            "credentials_inputs": {}
        })
        response.raise_for_status()
        return response.json()

# 使用示例
async def main():
    client = AutoGPTClient(
        base_url="http://localhost:8000",
        api_key="your-api-key"
    )
    
    # 创建图
    graph_data = {
        "name": "API创建的文本摘要流程",
        "description": "通过API创建的示例工作流",
        "nodes": [
            {
                "id": "text_input",
                "block_id": "text_input_block",
                "input_default": {},
                "metadata": {"position": {"x": 0, "y": 0}}
            },
            {
                "id": "summarize",
                "block_id": "llm_call_block", 
                "input_default": {
                    "model": "gpt-4",
                    "prompt": "请简洁地总结以下内容：\n\n{input_text}",
                    "max_tokens": 150
                },
                "metadata": {"position": {"x": 200, "y": 0}}
            }
        ],
        "links": [
            {
                "source_id": "text_input",
                "sink_id": "summarize", 
                "source_name": "text",
                "sink_name": "input_text"
            }
        ]
    }
    
    # 创建图
    graph = await client.create_graph(graph_data)
    print(f"创建图成功，ID: {graph['id']}")
    
    # 执行图
    execution = await client.execute_graph(
        graph_id=graph['id'],
        inputs={"text": "这是一段需要摘要的长文本..."}
    )
    print(f"执行开始，执行ID: {execution['id']}")

if __name__ == "__main__":
    asyncio.run(main())

graph TB
    subgraph "AutoGPT API架构"
        subgraph "客户端层 - Client Layer"
            WebUI["Web前端界面"]
            MobileApp["移动应用"]
            ThirdParty["第三方集成"]
            CLI["命令行工具"]
        end
        
        subgraph "API网关层 - API Gateway"
            RestAPI["REST API服务器"]
            ExternalAPI["外部API服务器"]
            WebSocketAPI["WebSocket API"]
        end
        
        subgraph "认证授权层 - Auth Layer"
            JWTAuth["JWT认证"]
            APIKeyAuth["API Key认证"]
            RateLimit["限流控制"]
            RBAC["角色权限控制"]
        end
        
        subgraph "业务逻辑层 - Business Logic"
            GraphMgmt["图管理"]
            ExecutionMgmt["执行管理"]
            UserMgmt["用户管理"]
            BlockMgmt["Block管理"]
        end
        
        subgraph "数据存储层 - Data Storage"
            PostgresDB[("PostgreSQL数据库")]
            RedisCache[("Redis缓存")]
            FileStorage[("文件存储")]
        end
    end
    
    %% 连接关系
    WebUI --> RestAPI
    MobileApp --> RestAPI
    ThirdParty --> ExternalAPI
    CLI --> RestAPI
    
    RestAPI --> JWTAuth
    ExternalAPI --> APIKeyAuth
    WebSocketAPI --> JWTAuth
    
    JWTAuth --> RateLimit
    APIKeyAuth --> RateLimit
    RateLimit --> RBAC
    
    RBAC --> GraphMgmt
    RBAC --> ExecutionMgmt
    RBAC --> UserMgmt
    RBAC --> BlockMgmt
    
    GraphMgmt --> PostgresDB
    ExecutionMgmt --> PostgresDB
    ExecutionMgmt --> RedisCache
    UserMgmt --> PostgresDB
    BlockMgmt --> FileStorage

# 文件: autogpt_platform/backend/backend/server/routers/v1.py
@v1_router.post(
    path="/graphs/{graph_id}/execute/{graph_version}",
    summary="Execute graph agent",
    tags=["graphs"],
    dependencies=[Security(requires_user)],
)
async def execute_graph(
    graph_id: str,  # 图的唯一标识符
    user_id: Annotated[str, Security(get_user_id)],  # 用户ID，从JWT中提取
    inputs: Annotated[dict[str, Any], Body(..., embed=True, default_factory=dict)],  # 执行输入数据
    credentials_inputs: Annotated[
        dict[str, CredentialsMetaInput], Body(..., embed=True, default_factory=dict)
    ],  # 凭据输入数据
    graph_version: Optional[int] = None,  # 图版本号，可选
    preset_id: Optional[str] = None,  # 预设ID，可选
) -> execution_db.GraphExecutionMeta:
    """
    执行工作流图的主要入口函数
    
    功能说明:
    1. 验证用户余额是否充足
    2. 调用执行工具函数创建图执行
    3. 记录执行指标和日志
    4. 处理各种异常情况
    
    参数说明:
    - graph_id: 要执行的图的唯一标识符
    - user_id: 执行用户的ID，用于权限验证和计费
    - inputs: 图执行的输入数据，键值对格式
    - credentials_inputs: 执行所需的凭据信息
    - graph_version: 指定执行的图版本，默认使用最新版本
    - preset_id: 预设配置ID，可以预定义输入参数
    
    返回值:
    - GraphExecutionMeta: 包含执行ID、状态、创建时间等元信息
    
    异常处理:
    - HTTPException(402): 余额不足时返回
    - HTTPException(400): 图验证失败时返回，包含详细错误信息
    - 其他异常: 记录日志后重新抛出
    """
    
    # 第一步：余额验证
    # 检查用户当前余额，确保有足够的积分执行图
    current_balance = await _user_credit_model.get_credits(user_id)
    if current_balance <= 0:
        raise HTTPException(
            status_code=402,  # Payment Required
            detail="Insufficient balance to execute the agent. Please top up your account.",
        )

    try:
        # 第二步：调用执行工具函数
        # 这是核心的图执行逻辑，创建执行实例并加入执行队列
        result = await execution_utils.add_graph_execution(
            graph_id=graph_id,
            user_id=user_id,
            inputs=inputs,
            preset_id=preset_id,
            graph_version=graph_version,
            graph_credentials_inputs=credentials_inputs,
        )
        
        # 第三步：记录成功指标
        # 用于监控和分析的指标记录
        record_graph_execution(graph_id=graph_id, status="success", user_id=user_id)
        record_graph_operation(operation="execute", status="success")
        
        return result
        
    except GraphValidationError as e:
        # 处理图验证错误
        # 这类错误通常是由于图结构不正确或节点配置有误
        record_graph_execution(
            graph_id=graph_id, status="validation_error", user_id=user_id
        )
        record_graph_operation(operation="execute", status="validation_error")
        
        # 返回结构化的验证错误信息，前端可以解析并显示
        raise HTTPException(
            status_code=400,
            detail={
                "type": "validation_error",
                "message": e.message,
                "node_errors": e.node_errors,  # 节点级别的错误详情
            },
        )
    except Exception:
        # 处理其他未预期的异常
        record_graph_execution(graph_id=graph_id, status="error", user_id=user_id)
        record_graph_operation(operation="execute", status="error")
        raise

sequenceDiagram
    participant Client as 客户端
    participant API as execute_graph API
    participant Auth as 认证模块
    participant Credit as 积分模块
    participant ExecUtils as execution_utils
    participant DB as 数据库
    participant Queue as 执行队列
    participant EM as ExecutionManager

    Client->>API: POST /api/graphs/{id}/execute
    API->>Auth: Security(requires_user)
    Auth->>Auth: 验证JWT令牌
    Auth-->>API: 返回用户ID
    
    API->>Credit: get_credits(user_id)
    Credit->>DB: 查询用户余额
    DB-->>Credit: 返回余额信息
    Credit-->>API: 返回当前余额
    
    alt 余额不足
        API-->>Client: HTTP 402 余额不足
    else 余额充足
        API->>ExecUtils: add_graph_execution(...)
        ExecUtils->>DB: 创建执行记录
        ExecUtils->>Queue: 加入执行队列
        Queue-->>EM: 通知有新执行任务
        ExecUtils-->>API: 返回执行元信息
        API-->>Client: 返回执行ID和状态
    end

# 文件: autogpt_platform/backend/backend/util/execution.py
async def add_graph_execution(
    graph_id: str,
    user_id: str,
    inputs: dict[str, Any],
    preset_id: Optional[str] = None,
    graph_version: Optional[int] = None,
    graph_credentials_inputs: Optional[dict[str, CredentialsMetaInput]] = None,
) -> execution_db.GraphExecutionMeta:
    """
    创建并启动图执行的核心函数
    
    功能说明:
    1. 验证图的存在性和用户权限
    2. 处理预设配置的合并
    3. 验证图结构和输入数据
    4. 创建执行记录并加入队列
    5. 发送实时通知
    
    参数说明:
    - graph_id: 图ID
    - user_id: 用户ID
    - inputs: 输入数据字典
    - preset_id: 预设ID，用于加载预定义配置
    - graph_version: 图版本，默认使用最新版本
    - graph_credentials_inputs: 凭据输入配置
    
    返回值:
    - GraphExecutionMeta: 执行元信息对象
    
    异常:
    - ValueError: 图不存在或用户无权限时抛出
    - GraphValidationError: 图验证失败时抛出
    """
    
    # 第一步：获取图定义
    # 根据图ID和版本获取图的完整定义
    graph = await graph_db.get_graph(
        graph_id=graph_id,
        version=graph_version,
        user_id=user_id,  # 用于权限验证
    )
    
    if not graph:
        raise ValueError(f"Graph #{graph_id} not found or access denied.")

    # 第二步：处理预设配置
    # 如果指定了预设ID，则加载预设配置并合并到输入中
    if preset_id:
        preset = await preset_db.get_preset(preset_id, user_id)
        if not preset:
            raise ValueError(f"Preset #{preset_id} not found.")
        
        # 合并预设的输入数据
        merged_inputs = {**preset.inputs, **inputs}
        
        # 合并预设的凭据配置
        if preset.credentials:
            merged_credentials = {**preset.credentials, **(graph_credentials_inputs or {})}
            graph_credentials_inputs = merged_credentials
        
        inputs = merged_inputs

    # 第三步：图验证
    # 验证图的结构完整性和输入数据的有效性
    try:
        validate_graph_structure(graph)
        validate_graph_inputs(graph, inputs)
    except Exception as e:
        raise GraphValidationError(
            message=f"Graph validation failed: {str(e)}",
            node_errors=getattr(e, 'node_errors', {})
        )

    # 第四步：创建执行记录
    # 在数据库中创建执行记录，状态为QUEUED
    graph_exec = await execution_db.create_graph_execution(
        graph_id=graph_id,
        graph_version=graph.version,
        user_id=user_id,
        inputs=inputs,
        credentials_inputs=graph_credentials_inputs or {},
        status=execution_db.ExecutionStatus.QUEUED,
    )

    # 第五步：加入执行队列
    # 创建执行队列条目并加入Redis队列
    execution_entry = GraphExecutionEntry(
        id=graph_exec.id,
        graph_id=graph_id,
        graph_version=graph.version,
        user_id=user_id,
        inputs=inputs,
        credentials_inputs=graph_credentials_inputs or {},
        created_at=datetime.utcnow(),
    )
    
    # 将执行条目加入Redis队列，ExecutionManager会从队列中取出并执行
    execution_queue = get_execution_queue()
    await execution_queue.put(execution_entry)
    
    logger.info(f"Graph execution {graph_exec.id} queued for user {user_id}")

    # 第六步：发送实时通知
    # 通过WebSocket向前端发送执行开始的通知
    await send_execution_update(
        user_id=user_id,
        graph_exec_id=graph_exec.id,
        status="queued",
        message="Execution queued successfully"
    )

    return graph_exec

@v1_router.post(
    path="/graphs/{graph_id}/executions/{graph_exec_id}/stop",
    summary="Stop graph execution",
    tags=["graphs"],
    dependencies=[Security(requires_user)],
)
async def stop_graph_run(
    graph_id: str,  # 图ID
    graph_exec_id: str,  # 执行ID
    user_id: Annotated[str, Security(get_user_id)]  # 用户ID
) -> execution_db.GraphExecutionMeta | None:
    """
    停止图执行的API端点
    
    功能说明:
    1. 验证用户权限
    2. 查找正在执行的图实例
    3. 发送停止信号
    4. 更新执行状态
    
    参数说明:
    - graph_id: 图的唯一标识符
    - graph_exec_id: 执行实例的唯一标识符
    - user_id: 请求用户的ID，用于权限验证
    
    返回值:
    - GraphExecutionMeta | None: 停止的执行信息，如果未找到则返回None
    """
    
    # 调用内部停止函数
    res = await _stop_graph_run(
        user_id=user_id,
        graph_id=graph_id,
        graph_exec_id=graph_exec_id,
    )
    
    # 如果没有找到可停止的执行，返回None
    if not res:
        return None
    
    # 返回第一个停止的执行信息
    return res[0]

async def _stop_graph_run(
    user_id: str,
    graph_id: Optional[str] = None,
    graph_exec_id: Optional[str] = None,
) -> list[execution_db.GraphExecutionMeta]:
    """
    内部停止图执行的实现函数
    
    功能说明:
    1. 查询符合条件的正在执行的图实例
    2. 批量发送停止信号
    3. 等待所有停止操作完成
    
    参数说明:
    - user_id: 用户ID，确保只能停止自己的执行
    - graph_id: 可选的图ID，用于过滤特定图的执行
    - graph_exec_id: 可选的执行ID，用于停止特定执行
    
    返回值:
    - list[GraphExecutionMeta]: 被停止的执行列表
    """
    
    # 第一步：查询可停止的执行
    # 只查询处于可停止状态的执行（排队中、运行中、未完成）
    graph_execs = await execution_db.get_graph_executions(
        user_id=user_id,
        graph_id=graph_id,
        graph_exec_id=graph_exec_id,
        statuses=[
            execution_db.ExecutionStatus.INCOMPLETE,  # 未完成
            execution_db.ExecutionStatus.QUEUED,     # 排队中
            execution_db.ExecutionStatus.RUNNING,    # 运行中
        ],
    )
    
    # 第二步：批量创建停止任务
    # 为每个找到的执行创建停止任务
    stopped_execs = [
        execution_utils.stop_graph_execution(
            graph_exec_id=exec.id, 
            user_id=user_id
        )
        for exec in graph_execs
    ]
    
    # 第三步：并发执行所有停止操作
    # 使用asyncio.gather并发执行，提高效率
    await asyncio.gather(*stopped_execs)
    
    return graph_execs

graph TB
    subgraph "AutoGPT平台整体架构"
        subgraph "前端层 - Frontend Layer"
            WebApp["React Web应用"]
            MobileApp["移动应用"]
            AdminPanel["管理后台"]
        end
        
        subgraph "API网关层 - API Gateway Layer"
            APIGateway["API网关"]
            LoadBalancer["负载均衡器"]
            RateLimit["限流服务"]
        end
        
        subgraph "应用服务层 - Application Service Layer"
            AuthService["认证服务"]
            GraphService["图管理服务"]
            ExecutionService["执行服务"]
            BlockService["Block服务"]
            UserService["用户服务"]
            NotificationService["通知服务"]
        end
        
        subgraph "执行引擎层 - Execution Engine Layer"
            ExecutionManager["执行管理器"]
            ExecutionProcessor["执行处理器"]
            NodeExecutor["节点执行器"]
            BlockRegistry["Block注册表"]
        end
        
        subgraph "数据层 - Data Layer"
            PostgresDB[("PostgreSQL主数据库")]
            RedisCache[("Redis缓存")]
            RedisQueue[("Redis消息队列")]
            FileStorage[("文件存储")]
            VectorDB[("向量数据库")]
        end
        
        subgraph "基础设施层 - Infrastructure Layer"
            Docker["Docker容器"]
            K8s["Kubernetes集群"]
            Monitoring["监控系统"]
            Logging["日志系统"]
        end
    end
    
    %% 连接关系
    WebApp --> APIGateway
    MobileApp --> APIGateway
    AdminPanel --> APIGateway
    
    APIGateway --> LoadBalancer
    LoadBalancer --> RateLimit
    
    RateLimit --> AuthService
    RateLimit --> GraphService
    RateLimit --> ExecutionService
    RateLimit --> BlockService
    RateLimit --> UserService
    
    ExecutionService --> ExecutionManager
    ExecutionManager --> ExecutionProcessor
    ExecutionProcessor --> NodeExecutor
    NodeExecutor --> BlockRegistry
    
    AuthService --> PostgresDB
    GraphService --> PostgresDB
    ExecutionService --> PostgresDB
    UserService --> PostgresDB
    
    ExecutionManager --> RedisQueue
    NotificationService --> RedisQueue
    
    BlockService --> FileStorage
    GraphService --> VectorDB
    
    ExecutionManager --> RedisCache
    AuthService --> RedisCache
    
    Docker --> K8s
    K8s --> Monitoring
    K8s --> Logging

graph TB
    subgraph "核心模块交互关系"
        subgraph "用户交互层"
            UI[Web界面]
            API[REST API]
            WS[WebSocket]
        end
        
        subgraph "业务逻辑层"
            GM[图管理模块]
            EM[执行管理模块]
            BM[Block管理模块]
            UM[用户管理模块]
            AM[认证模块]
        end
        
        subgraph "执行引擎层"
            EXE[执行引擎]
            PROC[执行处理器]
            NODE[节点执行器]
        end
        
        subgraph "数据存储层"
            DB[(数据库)]
            CACHE[(缓存)]
            QUEUE[(消息队列)]
        end
    end
    
    %% 用户交互
    UI --> API
    UI --> WS
    
    %% API路由
    API --> AM
    AM --> GM
    AM --> EM
    AM --> BM
    AM --> UM
    
    %% 执行流程
    EM --> EXE
    EXE --> PROC
    PROC --> NODE
    NODE --> BM
    
    %% 数据访问
    GM --> DB
    EM --> DB
    BM --> DB
    UM --> DB
    AM --> CACHE
    
    %% 异步通信
    EXE --> QUEUE
    PROC --> WS
    
    %% 反馈循环
    NODE --> EM
    PROC --> EM
    EXE --> API

flowchart TD
    subgraph "数据流架构"
        subgraph "输入层"
            UserInput[用户输入]
            APIInput[API输入]
            FileInput[文件输入]
        end
        
        subgraph "处理层"
            Validation[数据验证]
            Transformation[数据转换]
            Enrichment[数据增强]
        end
        
        subgraph "执行层"
            GraphExec[图执行]
            NodeExec[节点执行]
            BlockExec[Block执行]
        end
        
        subgraph "输出层"
            Results[执行结果]
            Notifications[通知消息]
            Logs[日志记录]
        end
        
        subgraph "存储层"
            TempStorage[临时存储]
            PersistentStorage[持久化存储]
            CacheStorage[缓存存储]
        end
    end
    
    %% 数据流向
    UserInput --> Validation
    APIInput --> Validation
    FileInput --> Validation
    
    Validation --> Transformation
    Transformation --> Enrichment
    
    Enrichment --> GraphExec
    GraphExec --> NodeExec
    NodeExec --> BlockExec
    
    BlockExec --> Results
    BlockExec --> Notifications
    BlockExec --> Logs
    
    %% 存储交互
    Enrichment --> TempStorage
    Results --> PersistentStorage
    BlockExec --> CacheStorage
    
    %% 反馈循环
    Results --> UserInput
    Logs --> Validation

classDiagram
    class ExecutionManager {
        -pool: ThreadPoolExecutor
        -active_runs: dict
        -cancel_events: dict
        +start()
        +stop()
        +run_execution_loop()
        +run_graph_execution()
        +cancel_execution()
    }
    
    class ExecutionProcessor {
        -db_client: DatabaseClient
        -creds_manager: CredentialsManager
        +on_graph_executor_start()
        +on_graph_execution()
        +on_node_execution()
    }
    
    class NodeExecutor {
        +execute_node()
        +validate_exec()
        +_enqueue_next_nodes()
        +block_usage_cost()
    }
    
    class GraphExecutionEntry {
        +id: str
        +graph_id: str
        +user_id: str
        +inputs: dict
        +created_at: datetime
    }
    
    class NodeExecutionEntry {
        +id: str
        +graph_exec_id: str
        +node_id: str
        +inputs: dict
        +user_context: UserContext
    }
    
    ExecutionManager --> ExecutionProcessor
    ExecutionProcessor --> NodeExecutor
    ExecutionManager --> GraphExecutionEntry
    ExecutionProcessor --> NodeExecutionEntry

sequenceDiagram
    participant Client as 客户端
    participant API as REST API
    participant EM as ExecutionManager
    participant EP as ExecutionProcessor
    participant NE as NodeExecutor
    participant DB as 数据库
    participant WS as WebSocket

    Client->>API: 提交执行请求
    API->>DB: 创建执行记录
    API->>EM: 加入执行队列
    
    EM->>EM: 从队列获取任务
    EM->>EP: 启动图执行
    
    EP->>DB: 获取图定义
    EP->>EP: 初始化节点队列
    
    loop 节点执行循环
        EP->>NE: 执行节点
        NE->>NE: 验证输入
        NE->>NE: 执行Block
        NE->>DB: 保存输出
        NE->>WS: 发送状态更新
        NE->>EP: 返回执行结果
        EP->>EP: 调度后续节点
    end
    
    EP->>DB: 更新执行状态
    EP->>WS: 发送完成通知
    EP-->>EM: 执行完成
    EM-->>API: 返回结果

classDiagram
    class Block {
        <<abstract>>
        +id: str
        +name: str
        +description: str
        +input_schema: BaseModel
        +output_schema: BaseModel
        +execute(input_data, **kwargs)*
    }
    
    class LLMBlock {
        +model: str
        +temperature: float
        +max_tokens: int
        +execute(input_data, **kwargs)
    }
    
    class TextProcessingBlock {
        +operation: str
        +execute(input_data, **kwargs)
    }
    
    class FileOperationBlock {
        +operation_type: str
        +execute(input_data, **kwargs)
    }
    
    class ConditionalBlock {
        +condition: str
        +execute(input_data, **kwargs)
    }
    
    class BlockRegistry {
        -blocks: dict
        +register_block(block)
        +get_block(block_id)
        +list_blocks()
    }
    
    Block <|-- LLMBlock
    Block <|-- TextProcessingBlock
    Block <|-- FileOperationBlock
    Block <|-- ConditionalBlock
    BlockRegistry --> Block

flowchart TD
    subgraph "认证授权流程"
        Start([开始])
        CheckToken{检查令牌}
        ValidateJWT{验证JWT}
        CheckAPIKey{检查API Key}
        ExtractUser[提取用户信息]
        CheckPermission{检查权限}
        AllowAccess[允许访问]
        DenyAccess[拒绝访问]
        End([结束])
    end
    
    Start --> CheckToken
    CheckToken -->|有JWT令牌| ValidateJWT
    CheckToken -->|有API Key| CheckAPIKey
    CheckToken -->|无令牌| DenyAccess
    
    ValidateJWT -->|有效| ExtractUser
    ValidateJWT -->|无效| DenyAccess
    
    CheckAPIKey -->|有效| ExtractUser
    CheckAPIKey -->|无效| DenyAccess
    
    ExtractUser --> CheckPermission
    CheckPermission -->|有权限| AllowAccess
    CheckPermission -->|无权限| DenyAccess
    
    AllowAccess --> End
    DenyAccess --> End

classDiagram
    class User {
        +id: str
        +email: str
        +name: str
        +role: str
        +created_at: datetime
        +last_login: datetime
        +timezone: str
        +credits: int
    }
    
    class Graph {
        +id: str
        +name: str
        +description: str
        +user_id: str
        +version: int
        +is_active: bool
        +created_at: datetime
        +updated_at: datetime
        +nodes: List[Node]
        +links: List[Link]
    }
    
    class Node {
        +id: str
        +graph_id: str
        +block_id: str
        +input_default: dict
        +metadata: dict
        +position: Position
    }
    
    class Link {
        +id: str
        +graph_id: str
        +source_id: str
        +sink_id: str
        +source_name: str
        +sink_name: str
    }
    
    class GraphExecution {
        +id: str
        +graph_id: str
        +user_id: str
        +status: ExecutionStatus
        +inputs: dict
        +outputs: dict
        +started_at: datetime
        +ended_at: datetime
        +error_message: str
    }
    
    class NodeExecution {
        +id: str
        +graph_execution_id: str
        +node_id: str
        +status: ExecutionStatus
        +inputs: dict
        +outputs: dict
        +started_at: datetime
        +ended_at: datetime
        +execution_time: float
    }
    
    class Block {
        +id: str
        +name: str
        +description: str
        +category: str
        +input_schema: dict
        +output_schema: dict
        +costs: List[BlockCost]
    }
    
    class APIKey {
        +id: str
        +user_id: str
        +name: str
        +key_hash: str
        +permissions: List[str]
        +is_active: bool
        +created_at: datetime
        +last_used: datetime
    }
    
    User ||--o{ Graph : owns
    User ||--o{ GraphExecution : executes
    User ||--o{ APIKey : has
    Graph ||--o{ Node : contains
    Graph ||--o{ Link : contains
    Graph ||--o{ GraphExecution : executed_as
    GraphExecution ||--o{ NodeExecution : contains
    Node ||--o{ NodeExecution : executed_as
    Block ||--o{ Node : instantiated_as

stateDiagram-v2
    [*] --> QUEUED : 创建执行
    QUEUED --> RUNNING : 开始执行
    QUEUED --> CANCELLED : 用户取消
    
    RUNNING --> COMPLETED : 执行成功
    RUNNING --> FAILED : 执行失败
    RUNNING --> CANCELLED : 用户取消
    
    COMPLETED --> [*]
    FAILED --> [*]
    CANCELLED --> [*]
    
    note right of QUEUED : 等待执行资源
    note right of RUNNING : 正在执行中
    note right of COMPLETED : 执行成功完成
    note right of FAILED : 执行失败
    note right of CANCELLED : 被用户取消

class GraphOptimizer:
    """图优化器 - 提供图结构优化建议和自动优化功能"""
    
    @staticmethod
    def analyze_graph_performance(graph: Graph) -> Dict[str, Any]:
        """
        分析图的性能特征
        
        Args:
            graph: 要分析的图对象
            
        Returns:
            包含性能分析结果的字典
        """
        analysis = {
            "node_count": len(graph.nodes),
            "link_count": len(graph.links),
            "max_depth": GraphOptimizer._calculate_max_depth(graph),
            "parallel_branches": GraphOptimizer._count_parallel_branches(graph),
            "bottlenecks": GraphOptimizer._identify_bottlenecks(graph),
            "optimization_suggestions": []
        }
        
        # 生成优化建议
        if analysis["node_count"] > 50:
            analysis["optimization_suggestions"].append(
                "考虑将大图拆分为多个子图以提高可维护性"
            )
        
        if analysis["max_depth"] > 10:
            analysis["optimization_suggestions"].append(
                "图的深度过深，考虑增加并行分支来减少执行时间"
            )
        
        if len(analysis["bottlenecks"]) > 0:
            analysis["optimization_suggestions"].append(
                f"发现性能瓶颈节点: {', '.join(analysis['bottlenecks'])}"
            )
        
        return analysis
    
    @staticmethod
    def _calculate_max_depth(graph: Graph) -> int:
        """计算图的最大深度"""
        # 构建邻接表
        adjacency = defaultdict(list)
        for link in graph.links:
            adjacency[link.source_id].append(link.sink_id)
        
        # 找到入口节点（没有输入的节点）
        all_nodes = {node.id for node in graph.nodes}
        sink_nodes = {link.sink_id for link in graph.links}
        entry_nodes = all_nodes - sink_nodes
        
        # DFS计算最大深度
        max_depth = 0
        visited = set()
        
        def dfs(node_id: str, depth: int):
            nonlocal max_depth
            if node_id in visited:
                return
            
            visited.add(node_id)
            max_depth = max(max_depth, depth)
            
            for next_node in adjacency[node_id]:
                dfs(next_node, depth + 1)
        
        for entry_node in entry_nodes:
            dfs(entry_node, 1)
        
        return max_depth

class ExecutionOptimizer:
    """执行优化器 - 提供执行性能优化策略"""
    
    @staticmethod
    def optimize_node_scheduling(nodes: List[Node]) -> List[List[Node]]:
        """
        优化节点调度，将可并行执行的节点分组
        
        Args:
            nodes: 节点列表
            
        Returns:
            按执行批次分组的节点列表
        """
        # 构建依赖关系图
        dependencies = ExecutionOptimizer._build_dependency_graph(nodes)
        
        # 拓扑排序确定执行顺序
        execution_batches = []
        remaining_nodes = set(node.id for node in nodes)
        
        while remaining_nodes:
            # 找到当前可执行的节点（没有未满足的依赖）
            ready_nodes = []
            for node_id in remaining_nodes:
                if not dependencies[node_id] or dependencies[node_id].isdisjoint(remaining_nodes):
                    ready_nodes.append(node_id)
            
            if not ready_nodes:
                # 检测到循环依赖
                raise ValueError("Circular dependency detected in graph")
            
            # 将就绪的节点加入当前批次
            batch_nodes = [node for node in nodes if node.id in ready_nodes]
            execution_batches.append(batch_nodes)
            
            # 从剩余节点中移除已调度的节点
            remaining_nodes -= set(ready_nodes)
        
        return execution_batches
    
    @staticmethod
    def estimate_execution_cost(graph: Graph, inputs: dict) -> Dict[str, float]:
        """
        估算图执行成本
        
        Args:
            graph: 要执行的图
            inputs: 输入数据
            
        Returns:
            包含成本估算的字典
        """
        cost_estimation = {
            "total_credits": 0.0,
            "estimated_time": 0.0,
            "node_costs": {},
            "critical_path_time": 0.0
        }
        
        for node in graph.nodes:
            block = get_block(node.block_id)
            if block:
                # 估算节点成本
                node_cost = ExecutionOptimizer._estimate_node_cost(block, node.input_default)
                node_time = ExecutionOptimizer._estimate_node_time(block, node.input_default)
                
                cost_estimation["node_costs"][node.id] = {
                    "credits": node_cost,
                    "time": node_time
                }
                cost_estimation["total_credits"] += node_cost
        
        # 计算关键路径时间
        cost_estimation["critical_path_time"] = ExecutionOptimizer._calculate_critical_path_time(
            graph, cost_estimation["node_costs"]
        )
        
        return cost_estimation

from enum import Enum
from typing import Optional, Dict, Any

class ErrorSeverity(Enum):
    """错误严重程度枚举"""
    LOW = "low"           # 轻微错误，不影响主流程
    MEDIUM = "medium"     # 中等错误，影响部分功能
    HIGH = "high"         # 严重错误，影响主要功能
    CRITICAL = "critical" # 致命错误，系统无法继续运行

class ErrorCategory(Enum):
    """错误类别枚举"""
    VALIDATION = "validation"     # 数据验证错误
    AUTHENTICATION = "auth"       # 认证授权错误
    RESOURCE = "resource"         # 资源不足错误
    NETWORK = "network"           # 网络连接错误
    SYSTEM = "system"             # 系统内部错误
    BUSINESS = "business"         # 业务逻辑错误

class AutoGPTError(Exception):
    """AutoGPT平台基础异常类"""
    
    def __init__(
        self,
        message: str,
        error_code: str,
        severity: ErrorSeverity = ErrorSeverity.MEDIUM,
        category: ErrorCategory = ErrorCategory.SYSTEM,
        details: Optional[Dict[str, Any]] = None,
        user_message: Optional[str] = None
    ):
        """
        初始化异常
        
        Args:
            message: 详细错误信息（用于日志）
            error_code: 错误代码（用于程序处理）
            severity: 错误严重程度
            category: 错误类别
            details: 错误详细信息
            user_message: 用户友好的错误信息
        """
        super().__init__(message)
        self.message = message
        self.error_code = error_code
        self.severity = severity
        self.category = category
        self.details = details or {}
        self.user_message = user_message or message
        self.timestamp = datetime.utcnow()
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式，用于API响应"""
        return {
            "error_code": self.error_code,
            "message": self.user_message,
            "severity": self.severity.value,
            "category": self.category.value,
            "details": self.details,
            "timestamp": self.timestamp.isoformat()
        }

class GraphValidationError(AutoGPTError):
    """图验证错误"""
    
    def __init__(self, message: str, node_errors: Dict[str, str] = None):
        super().__init__(
            message=message,
            error_code="GRAPH_VALIDATION_ERROR",
            severity=ErrorSeverity.HIGH,
            category=ErrorCategory.VALIDATION,
            details={"node_errors": node_errors or {}},
            user_message="工作流配置存在错误，请检查节点设置"
        )

class InsufficientCreditsError(AutoGPTError):
    """积分不足错误"""
    
    def __init__(self, required_credits: float, available_credits: float):
        super().__init__(
            message=f"Insufficient credits: required {required_credits}, available {available_credits}",
            error_code="INSUFFICIENT_CREDITS",
            severity=ErrorSeverity.HIGH,
            category=ErrorCategory.RESOURCE,
            details={
                "required_credits": required_credits,
                "available_credits": available_credits
            },
            user_message="账户余额不足，请充值后重试"
        )

class ErrorHandler:
    """统一错误处理器"""
    
    @staticmethod
    def handle_api_error(error: Exception, request_context: Dict[str, Any]) -> Dict[str, Any]:
        """
        处理API错误并返回标准化响应
        
        Args:
            error: 发生的异常
            request_context: 请求上下文信息
            
        Returns:
            标准化的错误响应
        """
        if isinstance(error, AutoGPTError):
            # 处理已知的业务异常
            ErrorHandler._log_error(error, request_context)
            return {
                "success": False,
                "error": error.to_dict()
            }
        
        elif isinstance(error, httpx.HTTPStatusError):
            # 处理HTTP错误
            autogpt_error = ErrorHandler._convert_http_error(error)
            ErrorHandler._log_error(autogpt_error, request_context)
            return {
                "success": False,
                "error": autogpt_error.to_dict()
            }
        
        else:
            # 处理未知异常
            system_error = AutoGPTError(
                message=str(error),
                error_code="SYSTEM_ERROR",
                severity=ErrorSeverity.CRITICAL,
                category=ErrorCategory.SYSTEM,
                user_message="系统内部错误，请稍后重试"
            )
            ErrorHandler._log_error(system_error, request_context)
            return {
                "success": False,
                "error": system_error.to_dict()
            }
    
    @staticmethod
    def _log_error(error: AutoGPTError, context: Dict[str, Any]):
        """记录错误日志"""
        log_data = {
            "error_code": error.error_code,
            "severity": error.severity.value,
            "category": error.category.value,
            "message": error.message,
            "details": error.details,
            "context": context
        }
        
        if error.severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]:
            logger.error("AutoGPT Error", extra=log_data)
        else:
            logger.warning("AutoGPT Warning", extra=log_data)

from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry
import time
from functools import wraps

class MetricsCollector:
    """指标收集器"""
    
    def __init__(self):
        self.registry = CollectorRegistry()
        
        # 执行相关指标
        self.graph_executions_total = Counter(
            'autogpt_graph_executions_total',
            'Total number of graph executions',
            ['status', 'user_id', 'graph_id'],
            registry=self.registry
        )
        
        self.graph_execution_duration = Histogram(
            'autogpt_graph_execution_duration_seconds',
            'Graph execution duration in seconds',
            ['graph_id', 'status'],
            registry=self.registry
        )
        
        self.active_executions = Gauge(
            'autogpt_active_executions',
            'Number of currently active executions',
            registry=self.registry
        )
        
        # Block执行指标
        self.block_executions_total = Counter(
            'autogpt_block_executions_total',
            'Total number of block executions',
            ['block_type', 'status'],
            registry=self.registry
        )
        
        self.block_execution_duration = Histogram(
            'autogpt_block_execution_duration_seconds',
            'Block execution duration in seconds',
            ['block_type'],
            registry=self.registry
        )
        
        # API相关指标
        self.api_requests_total = Counter(
            'autogpt_api_requests_total',
            'Total number of API requests',
            ['method', 'endpoint', 'status_code'],
            registry=self.registry
        )
        
        self.api_request_duration = Histogram(
            'autogpt_api_request_duration_seconds',
            'API request duration in seconds',
            ['method', 'endpoint'],
            registry=self.registry
        )
    
    def record_graph_execution(self, graph_id: str, user_id: str, status: str, duration: float = None):
        """记录图执行指标"""
        self.graph_executions_total.labels(
            status=status,
            user_id=user_id,
            graph_id=graph_id
        ).inc()
        
        if duration is not None:
            self.graph_execution_duration.labels(
                graph_id=graph_id,
                status=status
            ).observe(duration)
    
    def record_block_execution(self, block_type: str, status: str, duration: float):
        """记录Block执行指标"""
        self.block_executions_total.labels(
            block_type=block_type,
            status=status
        ).inc()
        
        self.block_execution_duration.labels(
            block_type=block_type
        ).observe(duration)
    
    def record_api_request(self, method: str, endpoint: str, status_code: int, duration: float):
        """记录API请求指标"""
        self.api_requests_total.labels(
            method=method,
            endpoint=endpoint,
            status_code=status_code
        ).inc()
        
        self.api_request_duration.labels(
            method=method,
            endpoint=endpoint
        ).observe(duration)

# 全局指标收集器实例
metrics_collector = MetricsCollector()

def monitor_execution_time(metric_name: str = None):
    """执行时间监控装饰器"""
    def decorator(func):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            start_time = time.time()
            try:
                result = await func(*args, **kwargs)
                duration = time.time() - start_time
                
                # 记录成功执行
                if metric_name:
                    getattr(metrics_collector, metric_name).observe(duration)
                
                return result
            except Exception as e:
                duration = time.time() - start_time
                
                # 记录失败执行
                if metric_name:
                    getattr(metrics_collector, metric_name).observe(duration)
                
                raise
        
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            start_time = time.time()
            try:
                result = func(*args, **kwargs)
                duration = time.time() - start_time
                
                if metric_name:
                    getattr(metrics_collector, metric_name).observe(duration)
                
                return result
            except Exception as e:
                duration = time.time() - start_time
                
                if metric_name:
                    getattr(metrics_collector, metric_name).observe(duration)
                
                raise
        
        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    
    return decorator

import re
from typing import Any, Dict, List
from pydantic import BaseModel, validator
import bleach

class InputSanitizer:
    """输入数据清理器"""
    
    # 允许的HTML标签（用于富文本内容）
    ALLOWED_HTML_TAGS = [
        'p', 'br', 'strong', 'em', 'u', 'ol', 'ul', 'li',
        'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'blockquote'
    ]
    
    # 允许的HTML属性
    ALLOWED_HTML_ATTRIBUTES = {
        '*': ['class'],
        'a': ['href', 'title'],
        'img': ['src', 'alt', 'width', 'height']
    }
    
    @staticmethod
    def sanitize_html(html_content: str) -> str:
        """清理HTML内容，移除潜在的XSS攻击代码"""
        if not html_content:
            return ""
        
        return bleach.clean(
            html_content,
            tags=InputSanitizer.ALLOWED_HTML_TAGS,
            attributes=InputSanitizer.ALLOWED_HTML_ATTRIBUTES,
            strip=True
        )
    
    @staticmethod
    def sanitize_filename(filename: str) -> str:
        """清理文件名，移除危险字符"""
        if not filename:
            return ""
        
        # 移除路径遍历字符
        filename = filename.replace('..', '').replace('/', '').replace('\\', '')
        
        # 只保留字母、数字、点、下划线和连字符
        filename = re.sub(r'[^\w\-_\.]', '', filename)
        
        # 限制长度
        return filename[:255]
    
    @staticmethod
    def validate_json_structure(data: Any, max_depth: int = 10, max_keys: int = 1000) -> bool:
        """验证JSON结构，防止过深嵌套和过多键值对"""
        
        def count_items(obj, current_depth=0):
            if current_depth > max_depth:
                raise ValueError("JSON structure too deep")
            
            item_count = 0
            
            if isinstance(obj, dict):
                item_count += len(obj)
                if item_count > max_keys:
                    raise ValueError("Too many keys in JSON")
                
                for value in obj.values():
                    item_count += count_items(value, current_depth + 1)
            
            elif isinstance(obj, list):
                item_count += len(obj)
                if item_count > max_keys:
                    raise ValueError("Too many items in JSON")
                
                for item in obj:
                    item_count += count_items(item, current_depth + 1)
            
            return item_count
        
        try:
            count_items(data)
            return True
        except ValueError:
            return False

class SecureGraphInput(BaseModel):
    """安全的图输入数据模型"""
    
    name: str
    description: str = ""
    inputs: Dict[str, Any]
    
    @validator('name')
    def validate_name(cls, v):
        """验证图名称"""
        if not v or len(v.strip()) == 0:
            raise ValueError("Graph name cannot be empty")
        
        if len(v) > 100:
            raise ValueError("Graph name too long")
        
        # 移除HTML标签
        clean_name = InputSanitizer.sanitize_html(v)
        return clean_name.strip()
    
    @validator('description')
    def validate_description(cls, v):
        """验证图描述"""
        if len(v) > 1000:
            raise ValueError("Description too long")
        
        return InputSanitizer.sanitize_html(v)
    
    @validator('inputs')
    def validate_inputs(cls, v):
        """验证输入数据"""
        if not InputSanitizer.validate_json_structure(v):
            raise ValueError("Invalid input structure")
        
        return v

class RateLimitManager:
    """速率限制管理器"""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    async def check_rate_limit(
        self,
        key: str,
        limit: int,
        window: int = 3600,
        identifier: str = "default"
    ) -> Dict[str, Any]:
        """
        检查速率限制
        
        Args:
            key: 限制键（如用户ID、IP地址）
            limit: 限制次数
            window: 时间窗口（秒）
            identifier: 限制标识符
            
        Returns:
            包含限制状态的字典
        """
        rate_key = f"rate_limit:{identifier}:{key}"
        current_time = int(time.time())
        window_start = current_time - window
        
        # 使用Redis pipeline提高性能
        pipe = self.redis.pipeline()
        
        # 移除过期的记录
        pipe.zremrangebyscore(rate_key, 0, window_start)
        
        # 添加当前请求
        pipe.zadd(rate_key, {str(current_time): current_time})
        
        # 获取当前窗口内的请求数
        pipe.zcount(rate_key, window_start, current_time)
        
        # 设置键的过期时间
        pipe.expire(rate_key, window)
        
        results = await pipe.execute()
        current_requests = results[2]
        
        is_allowed = current_requests <= limit
        remaining = max(0, limit - current_requests)
        
        return {
            "allowed": is_allowed,
            "limit": limit,
            "remaining": remaining,
            "reset_time": current_time + window,
            "current_requests": current_requests
        }