企业AI工具平台架构设计：拥抱快速变化的AI生态

┌─────────────────────────────────────────────────────────────────────────┐
│                          分层架构                                        │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │ Layer 1: 持久化层 (数据资产)                                      │   │
│  │   PostgreSQL + 对象存储                                          │   │
│  │   • 用户数据    • Agent配置    • 知识库原文    • 对话历史        │   │
│  │   ★ 永久保留，跨版本迁移，跨供应商迁移                              │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                    │                                     │
│                                    ▼ (可重建)                            │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │ Layer 2: 索引层 (可重建)                                         │   │
│  │   向量数据库 + 搜索引擎                                           │   │
│  │   • 向量索引    • 倒排索引    • 全文索引                          │   │
│  │   ← 从 Layer 1 可以100%重建                                      │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                    │                                     │
│                                    ▼ (可替换)                            │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │ Layer 3: Agent工具层                                             │   │
│  │   从注册中心动态加载                                             │   │
│  │   • Claude Code    • OpenCAW    • OpenCode    • Custom...       │   │
│  │   ← 随时可替换，配置不变                                          │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                    │                                     │
│                                    ▼ (可扩展)                            │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │ Layer 4: AI模型层                                               │   │
│  │   适配器模式                                                     │   │
│  │   • OpenAI    • Claude    • GLM    • Qwen    • DeepSeek...      │   │
│  │   ← 新模型接入只需实现适配器                                      │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────┐
│                                                                         │
│   不怕换模型         不怕换Agent        不怕换向量数据库                 │
│   ─────────         ─────────         ──────────────                     │
│   适配器模式         注册中心           索引重建服务                     │
│   接口不变           热插拔             从Layer 1重建                     │
│                                                                         │
│   配置在DB           配置在DB           原文在PostgreSQL                 │
│   模型随便换         Agent随便换        向量随便换                       │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────┐
│                           用户层 (User Layer)                            │
│                    Web Console │ CLI │ API Gateway │ SDK │
├─────────────────────────────────────────────────────────────────────────┤
│                        业务应用层 (App Layer)                           │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │
│  │   智能体     │  │   工作流     │  │   知识库     │  │   插件市场   │ │
│  │   Agent │  │   Workflow   │  │   Knowledge  │  │   Marketplace│ │
│  └──────────────┘  └──────────────┘  └──────────────┘  └──────────────┘ │
├─────────────────────────────────────────────────────────────────────────┤
│                         编排调度层 (Orchestration)                      │
│      Agent Runtime │ Workflow Engine │ Task Scheduler │ Event Bus │
├─────────────────────────────────────────────────────────────────────────┤
│                         核心服务层 (Core Services)                      │
│   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐   │
│   │  用户管理    │  │  容器管理    │  │  项目管理    │  │  权限管理    │   │
│   │  User Mgmt │  │  Tenant │  │  Project    │  │  Permission │   │
│   └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘   │
├─────────────────────────────────────────────────────────────────────────┤
│                         集成层 (Integration Layer)                      │
│   ┌─────────────────────────────────────────────────────────────────┐   │
│   │                     Adapter Registry                             │   │
│   │  ┌────────┐ ┌────────┐┌────────┐ ┌────────┐ ┌────────┐       │   │
│   │  │ OpenAI │ │Claude │ │ Gemini │ │ 智谱   │ │ 通义   │ ... │   │
│   │  │Adapter │ │Adapter │ │Adapter │ │Adapter │ │Adapter │       │   │
│   │  └────────┘ └────────┘ └────────┘ └────────┘ └────────┘       │   │
│   └─────────────────────────────────────────────────────────────────┘   │
├─────────────────────────────────────────────────────────────────────────┤
│                         外部AI服务 (External AI Services)                │
│      OpenAI │ Anthropic │ Google │ 智谱AI │ 阿里云 │ 百度 │ ... │
└─────────────────────────────────────────────────────────────────────────┘

@dataclass
class Agent:
    id: str
    name: str
    description: str
    model_config: ModelConfig
    
    # 组成要素
    instructions: str          # 系统提示词
    tools: list[str]          # 关联的工具
    knowledge_ids: list[str]  # 关联的知识库
    
    # 元数据
    created_by: str
    project_id: str
    created_at: datetime
    updated_at: datetime
    
    # 分享与激励
    is_public: bool           # 是否公开分享
    likes: int = 0            # 点赞数
    uses: int = 0             # 使用次数
    
    # 版本控制
    version: int = 1
    version_history: list[dict] = field(default_factory=list)

@dataclass
class Workflow:
    id: str
    name: str
    description: str
    
    # 工作流定义
    nodes: list[WorkflowNode]
    edges: list[WorkflowEdge]
    
    # 执行配置
    trigger_type: TriggerType  # manual/scheduled/event
    trigger_config: dict
    timeout: int = 300         # 超时时间(秒)
    
    # 元数据
    created_by: str
    project_id: str
    is_public: bool
    likes: int = 0

@dataclass
class WorkflowNode(BaseModel):
    id: str
    type: NodeType
    config: dict
    position: tuple[int, int]
    
@dataclass  
class WorkflowEdge(BaseModel):
    id: str
    source: str
    target: str
    condition: str | None     # 条件判断
    label: str | None

┌──────────────────────────────────────────────────────────────┐
│                    工作流节点类型                              │
├──────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐        │
│  │  开始    │  │  Agent  │  │  工具   │  │  条件   │        │
│  │  Start  │  │         │  │  Tool   │  │ Condition│        │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘        │
│       │            │            │            │               │
│       ▼            ▼            ▼            ▼               │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐        │
│  │  输入    │  │  LLM    │  │  搜索   │  │  分支   │        │
│  │  Input  │  │  Call   │  │  Search │  │  Branch │        │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘        │
│       │            │            │            │               │
│       ▼            ▼            ▼            ▼               │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐        │
│  │  输出    │  │  知识库 │  │  脚本   │  │  循环   │        │
│  │  Output │  │  RAG    │  │  Script │  │  Loop   │        │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘        │
│       │            │            │            │               │
│       └────────────┴────────────┴────────────┘               │
│                         ▼                                     │
│                  ┌─────────┐                                 │
│                  │   结束   │                                 │
│                  │   End   │                                 │
│                  └─────────┘                                 │
└──────────────────────────────────────────────────────────────┘

@dataclass
class Plugin:
    id: str
    name: str
    version: str
    author: str
    description: str
    
    # 插件类型
    type: PluginType  # model/tool/memory/middleware
    
    # 能力定义
    capabilities: list[str]
    
    # 配置
    config_schema: dict
    default_config: dict
    
    # 分享与激励
    downloads: int = 0
    rating: float = 0.0
    reviews: list[Review] = field(default_factory=list)

@dataclass
class User:
    id: str
    name: str
    email: str
    phone: str | None
    avatar: str | None
    
    # 组织关系
    tenant_id: str           # 所属容器
    department: str          # 部门
    position: str           # 职位
    
    # 角色与权限
    role: UserRole           # admin/editor/viewer
    
    # 贡献积分
    contribution_score: int = 0
    badges: list[str] = field(default_factory=list)
    
    # 时间戳
    created_at: datetime
    last_login_at: datetime

class UserRole(Enum):
    TENANT_ADMIN = "tenant_admin"    # 容器管理员
    PROJECT_ADMIN = "project_admin" # 项目管理员
    EDITOR = "editor"               # 编辑者
    VIEWER = "viewer"              # 查看者

@dataclass
class Tenant:
    id: str
    name: str                      # 容器名称，如"研发中心"
    code: str                      # 唯一标识，如"rd"
    
    # 配置
    config: TenantConfig
    
    # 配额
    quotas: TenantQuotas
    
    # 成员
    member_count: int
    project_count: int
    
    # 状态
    status: TenantStatus = TenantStatus.ACTIVE
    created_at: datetime

@dataclass
class TenantConfig:
    # AI模型配置
    allowed_providers: list[str]   # 允许使用的AI提供商
    default_provider: str         # 默认提供商
    model_restrictions: dict       # 模型限制
    
    # 功能开关
    features: dict                 # 功能开关
    
    # 安全配置
    allowed_ips: list[str]         # IP白名单
    data_retention_days: int       # 数据保留天数

@dataclass
class TenantQuotas:
    max_users: int                 # 最大用户数
    max_projects: int              # 最大项目数
    max_agents: int                # 最大智能体数
    max_workflows: int             # 最大工作流数
    max_storage_gb: int             # 最大存储(GB)
    monthly_token_limit: int       # 月度Token限额

@dataclass
class Project:
    id: str
    name: str
    tenant_id: str                 # 所属容器
    code: str                      # 项目标识
    
    description: str
    icon: str | None
    
    # 成员
    owner_id: str                  # 项目负责人
    members: list[ProjectMember]
    
    # 配置
    config: ProjectConfig
    
    # 统计
    agent_count: int = 0
    workflow_count: int = 0
    knowledge_count: int = 0
    
    created_at: datetime
    updated_at: datetime

@dataclass
class ProjectMember:
    user_id: str
    role: ProjectRole              # owner/admin/member
    joined_at: datetime

class PermissionMatrix:
    """权限矩阵"""
    
    PERMISSIONS = {
        # 用户管理
        "user:create": ["tenant_admin"],
        "user:invite": ["tenant_admin"],
        "user:remove": ["tenant_admin"],
        
        # 容器管理
        "tenant:view": ["tenant_admin"],
        "tenant:edit": ["tenant_admin"],
        "tenant:quota": ["tenant_admin"],
        
        # 项目管理
        "project:create": ["tenant_admin", "project_admin"],
        "project:view": ["tenant_admin", "project_admin", "member"],
        "project:edit": ["tenant_admin", "project_admin"],
        "project:delete": ["tenant_admin", "project_admin"],
        
        # 智能体管理
        "agent:create": ["tenant_admin", "project_admin", "editor"],
        "agent:view": ["tenant_admin", "project_admin", "editor", "viewer"],
        "agent:edit": ["tenant_admin", "project_admin", "editor"],
        "agent:delete": ["tenant_admin", "project_admin", "editor"],
        "agent:publish": ["tenant_admin", "project_admin", "editor"],
        
        # 工作流管理
        "workflow:create": ["tenant_admin", "project_admin", "editor"],
        "workflow:view": ["tenant_admin", "project_admin", "editor", "viewer"],
        "workflow:edit": ["tenant_admin", "project_admin", "editor"],
        "workflow:delete": ["tenant_admin", "project_admin", "editor"],
        
        # 知识库管理
        "knowledge:create": ["tenant_admin", "project_admin", "editor"],
        "knowledge:view": ["tenant_admin", "project_admin", "editor", "viewer"],
        "knowledge:edit": ["tenant_admin", "project_admin", "editor"],
        "knowledge:delete": ["tenant_admin", "project_admin", "editor"],
    }
    
    @classmethod
    def check(cls, role: UserRole, permission: str) -> bool:
        return role.value in cls.PERMISSIONS.get(permission, [])

class AIProvider(Protocol):
    """所有AI提供者必须实现的接口"""
    
    @property
    def name(self) -> str:
        """提供商名称"""
        ...
    
    @property
    def supported_models(self) -> list[str]:
        """支持的模型列表"""
        ...
    
    async def chat_completion(
        self,
        messages: list[Message],
        model: str,
        temperature: float = 0.7,
        max_tokens: int | None = None,
        tools: list[Tool] | None = None,
        **kwargs
    ) -> ChatResponse:
        """聊天补全"""
        ...
    
    async def embeddings(
        self,
        texts: list[str],
        model: str = "embeddings",
        **kwargs
    ) -> EmbeddingResponse:
        """文本嵌入"""
        ...
    
    async def images_generate(
        self,
        prompt: str,
        model: str,
        size: str = "1024x1024",
        **kwargs
    ) -> ImageResponse:
        """图像生成"""
        ...
    
    async def models_list(self) -> list[ModelInfo]:
        """获取可用模型列表"""
        ...
    
    async def health_check(self) -> bool:
        """健康检查"""
        ...

class OpenAIAdapter:
    """OpenAI适配器"""
    
    def __init__(
        self,
        api_key: str,
        api_base: str = "https://api.openai.com/v1",
        timeout: int = 60
    ):
        self.api_key = api_key
        self.api_base = api_base
        self.client = OpenAIClient(api_key, api_base, timeout)
    
    @property
    def name(self) -> str:
        return "openai"
    
    @property
    def supported_models(self) -> list[str]:
        return [
            "gpt-4o",
            "gpt-4o-mini", 
            "gpt-4-turbo",
            "gpt-3.5-turbo",
            "text-embedding-3-small",
            "text-embedding-3-large",
            "dall-e-3",
            "dall-e-2"
        ]
    
    async def chat_completion(
        self,
        messages: list[Message],
        model: str = "gpt-4o",
        **kwargs
    ) -> ChatResponse:
        response = await self.client.chat.completions.create(
            model=model,
            messages=[m.to_dict() for m in messages],
            **kwargs
        )
        return ChatResponse.from_openai(response)
    
    async def embeddings(
        self,
        texts: list[str],
        model: str = "text-embedding-3-small"
    ) -> EmbeddingResponse:
        response = await self.client.embeddings.create(
            model=model,
            input=texts
        )
        return EmbeddingResponse.from_openai(response)


class ClaudeAdapter:
    """Anthropic Claude适配器"""
    
    def __init__(
        self,
        api_key: str,
        api_base: str = "https://api.anthropic.com/v1",
        timeout: int = 60
    ):
        self.api_key = api_key
        self.api_base = api_base
        self.client = AnthropicClient(api_key, api_base, timeout)
    
    @property
    def name(self) -> str:
        return "anthropic"
    
    @property
    def supported_models(self) -> list[str]:
        return [
            "claude-sonnet-4-20250514",
            "claude-3-5-sonnet-latest",
            "claude-3-opus-latest",
            "claude-3-haiku-latest"
        ]
    
    async def chat_completion(
        self,
        messages: list[Message],
        model: str = "claude-sonnet-4-20250514",
        **kwargs
    ) -> ChatResponse:
        response = await self.client.messages.create(
            model=model,
            messages=[m.to_anthropic_format() for m in messages],
            **kwargs
        )
        return ChatResponse.from_anthropic(response)


class DeepSeekAdapter:
    """DeepSeek适配器 - 演示新模型接入的简易性"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.client = OpenAICompatibleClient(
            api_key=api_key,
            api_base="https://api.deepseek.com/v1"
        )
    
    @property
    def name(self) -> str:
        return "deepseek"
    
    @property
    def supported_models(self) -> list[str]:
        return ["deepseek-chat", "deepseek-coder"]
    
    async def chat_completion(self, messages, model="deepseek-chat", **kwargs):
        response = await self.client.chat.completions.create(
            model=model,
            messages=[m.to_dict() for m in messages],
            **kwargs
        )
        return ChatResponse.from_openai(response)

class AdapterRegistry:
    """适配器注册中心 - 核心扩展机制"""
    
    def __init__(self):
        self._adapters: dict[str, AIProvider] = {}
        self._metadata: dict[str, AdapterMetadata] = {}
    
    def register(
        self,
        name: str,
        adapter: AIProvider,
        metadata: AdapterMetadata | None = None
    ):
        """注册适配器"""
        self._adapters[name] = adapter
        self._metadata[name] = metadata or AdapterMetadata(
            name=name,
            version="1.0.0",
            capabilities=["chat", "embeddings"]
        )
        logger.info(f"Registered adapter: {name}")
    
    def unregister(self, name: str):
        """注销适配器"""
        if name in self._adapters:
            del self._adapters[name]
            del self._metadata[name]
            logger.info(f"Unregistered adapter: {name}")
    
    def get(self, name: str) -> AIProvider | None:
        """获取适配器"""
        return self._adapters.get(name)
    
    def list_adapters(self) -> list[str]:
        """列出所有适配器"""
        return list(self._adapters.keys())
    
    def list_by_capability(self, capability: str) -> list[str]:
        """按能力筛选适配器"""
        return [
            name for name, meta in self._metadata.items()
            if capability in meta.capabilities
        ]
    
    async def health_check_all(self) -> dict[str, bool]:
        """检查所有适配器健康状态"""
        results = {}
        for name, adapter in self._adapters.items():
            try:
                results[name] = await adapter.health_check()
            except Exception as e:
                logger.error(f"Health check failed for {name}: {e}")
                results[name] = False
        return results


@dataclass
class AdapterMetadata:
    name: str
    version: str
    capabilities: list[str]
    description: str = ""
    author: str = ""
    docs_url: str = ""

class ToolCapability(Protocol):
    """工具能力抽象"""
    
    @property
    def name(self) -> str:
        ...
    
    @property
    def description(self) -> str:
        ...
    
    def get_schema(self) -> ToolSchema:
        """获取工具定义schema，用于LLM function calling"""
        ...
    
    async def execute(
        self,
        context: ExecutionContext,
        params: dict
    ) -> ToolResult:
        """执行工具"""
        ...


class ToolType(Enum):
    """工具类型枚举"""
    RETRIEVER = "retriever"           # 知识检索
    WEB_SEARCH = "web_search"          # 网络搜索
    CALCULATOR = "calculator"         # 计算器
    CODE_INTERPRETER = "code"         # 代码执行
    FILE_SYSTEM = "filesystem"        # 文件操作
    DATABASE = "database"             # 数据库查询
    API_CALL = "api_call"             # API调用
    IMAGE_PROCESSOR = "image"         # 图像处理
    DOCUMENT_PARSER = "document"      # 文档解析
    CUSTOM = "custom"                 # 自定义工具


class ToolRegistry:
    """工具注册中心"""
    
    def __init__(self):
        self._tools: dict[str, ToolCapability] = {}
        self._by_type: dict[ToolType, list[str]] = defaultdict(list)
    
    def register(self, tool: ToolCapability, types: list[ToolType] = None):
        """注册工具"""
        self._tools[tool.name] = tool
        for tool_type in (types or [ToolType.CUSTOM]):
            self._by_type[tool_type].append(tool.name)
        logger.info(f"Registered tool: {tool.name}")
    
    def execute(
        self,
        tool_name: str,
        params: dict,
        context: ExecutionContext
    ) -> ToolResult:
        """执行工具"""
        tool = self._tools.get(tool_name)
        if not tool:
            raise ToolNotFoundError(f"Tool not found: {tool_name}")
        return await tool.execute(context, params)
    
    def list_by_type(self, tool_type: ToolType) -> list[ToolCapability]:
        """按类型列出工具"""
        tool_names = self._by_type.get(tool_type, [])
        return [self._tools[name] for name in tool_names]

from abc import ABC, abstractmethod
from typing import Protocol, Any
from dataclasses import dataclass, field
from enum import Enum
import asyncio

class AgentFramework(Enum):
    """支持的Agent框架"""
    CLAUDE_CODE = "claude_code"
    OPENCODE = "opencode"
    CODEX = "codex"
    DEVIN = "devin"
    OPENCAW = "opencaw"
    SWE_AGENT = "swe_agent"
    CUSTOM = "custom"  # 支持自定义框架

@dataclass
class AgentExecutionRequest:
    """Agent执行请求"""
    task: str                          # 任务描述
    framework: AgentFramework          # 使用的框架
    working_dir: str # 工作目录
    timeout: int = 300                 # 超时时间(秒)
    allowed_tools: list[str] | None = None  # 允许的工具
    extra_config: dict = field(default_factory=dict)  # 框架特定配置

@dataclass
class AgentExecutionResponse:
    """Agent执行响应"""
    success: bool
    result: str | None
    error: str | None
    session_id: str | None
    duration_ms: int
    tokens_used: dict = field(default_factory=dict)
    metadata: dict = field(default_factory=dict)

@dataclass
class AgentFrameworkMetadata:
    """框架元数据"""
    name: str
    version: str
    capabilities: list[str]           # ["code_generation", "code_review", ...]
    supported_languages: list[str]     # 支持的编程语言
    requires_permissions: list[str]    # 需要的权限
    config_schema: dict                # 配置schema
    auth_method: str                  # 认证方式: "api_key", "oauth", "cli_auth"

class ExternalAgentAdapter(ABC):
    """外部Agent框架适配器基类"""
    
    def __init__(self, config: dict):
        self.config = config
    
    @abstractmethod
    async def execute(self, request: AgentExecutionRequest) -> AgentExecutionResponse:
        """执行Agent任务"""
        pass
    
    @abstractmethod
    async def health_check(self) -> bool:
        """健康检查"""
        pass
    
    @abstractmethod
    async def get_status(self) -> dict:
        """获取Agent状态"""
        pass
    
    def get_metadata(self) -> AgentFrameworkMetadata:
        """获取框架元数据"""
        raise NotImplementedError

import subprocess
import json
import asyncio
from pathlib import Path

class ClaudeCodeAdapter(ExternalAgentAdapter):
    """Claude Code适配器"""
    
    def __init__(self, config: dict):
        super().__init__(config)
        self.cli_path = config.get("cli_path", "claude")
        self.api_key = config.get("api_key")  # 可选，使用ANTHROPIC_API_KEY
        self.default_model = config.get("default_model", "sonnet")
    
    def get_metadata(self) -> AgentFrameworkMetadata:
        return AgentFrameworkMetadata(
            name="Claude Code",
            version="2.x",
            capabilities=[
                "code_generation",
                "code_review",
                "refactoring",
                "bug_fixing",
                "test_writing",
                "git_operations"
            ],
            supported_languages=[
                "python", "javascript", "typescript", "go", "rust",
                "java", "c", "cpp", "csharp", "ruby", "php"
            ],
            requires_permissions=["file_read", "file_write", "bash"],
            config_schema={
                "cli_path": {"type": "string", "default": "claude"},
                "default_model": {"type": "enum", "values": ["sonnet", "opus", "haiku"]},
                "max_turns": {"type": "int", "default": 10}
            },
            auth_method="api_key"
        )
    
    async def execute(self, request: AgentExecutionRequest) -> AgentExecutionResponse:
        """执行Claude Code任务"""
        start_time = asyncio.get_event_loop().time()
        
        # 构建命令
        cmd = [
            self.cli_path,
            "-p", request.task,
            "--model", request.extra_config.get("model", self.default_model),
            "--output-format", "json",
            "--max-turns", str(request.extra_config.get("max_turns", 10))
        ]
        
        # 处理工具限制
        if request.allowed_tools:
            cmd.extend(["--allowedTools", ",".join(request.allowed_tools)])
        
        # 权限绕过（生产环境需谨慎）
        if request.extra_config.get("skip_permissions"):
            cmd.append("--dangerously-skip-permissions")
        
        try:
            result = subprocess.run(
                cmd,
                cwd=request.working_dir,
                capture_output=True,
                text=True,
                timeout=request.timeout
            )
            
            duration_ms = int((asyncio.get_event_loop().time() - start_time) * 1000)
            
            if result.returncode == 0:
                # 解析JSON输出
                output = json.loads(result.stdout)
                return AgentExecutionResponse(
                    success=True,
                    result=output.get("result", ""),
                    session_id=output.get("session_id"),
                    duration_ms=duration_ms,
                    tokens_used=output.get("usage", {})
                )
            else:
                return AgentExecutionResponse(
                    success=False,
                    error=result.stderr,
                    duration_ms=duration_ms
                )
        except subprocess.TimeoutExpired:
            return AgentExecutionResponse(
                success=False,
                error=f"Task timeout after {request.timeout}s",
                duration_ms=request.timeout * 1000
            )
        except Exception as e:
            return AgentExecutionResponse(
                success=False,
                error=str(e),
                duration_ms=int((asyncio.get_event_loop().time() - start_time) * 1000)
            )
    
    async def health_check(self) -> bool:
        """检查Claude Code是否可用"""
        try:
            result = subprocess.run(
                [self.cli_path, "--version"],
                capture_output=True,
                text=True,
                timeout=5
            )
            return result.returncode == 0
        except Exception:
            return False
    
    async def get_status(self) -> dict:
        """获取Claude Code状态"""
        try:
            result = subprocess.run(
                [self.cli_path, "auth", "status", "--text"],
                capture_output=True,
                text=True,
                timeout=10
            )
            return {
                "installed": True,
                "authenticated": "logged in" in result.stdout.lower(),
                "version": result.stdout.split("\n")[0] if result.stdout else "unknown"
            }
        except Exception as e:
            return {"installed": False, "error": str(e)}

class OpenCAWAdapter(ExternalAgentAdapter):
    """OpenCAW适配器 - 国产开源Agent框架"""
    
    def __init__(self, config: dict):
        super().__init__(config)
        self.api_base = config.get("api_base", "http://localhost:8080")
        self.api_key = config.get("api_key")
        self.ws_endpoint = config.get("ws_endpoint")
    
    def get_metadata(self) -> AgentFrameworkMetadata:
        return AgentFrameworkMetadata(
            name="OpenCAW",
            version="1.x",
            capabilities=[
                "code_generation",
                "code_review",
                "mcp_integration",
                "multi_agent"
            ],
            supported_languages=["python", "javascript", "typescript", "go"],
            requires_permissions=["file_read", "file_write", "bash", "network"],
            config_schema={
                "api_base": {"type": "string"},
                "api_key": {"type": "string"},
                "ws_endpoint": {"type": "string"}
            },
            auth_method="api_key"
        )
    
    async def execute(self, request: AgentExecutionRequest) -> AgentExecutionResponse:
        """通过HTTP API执行OpenCAW任务"""
        import aiohttp
        
        start_time = asyncio.get_event_loop().time()
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "task": request.task,
            "working_dir": request.working_dir,
            "timeout": request.timeout,
            "config": request.extra_config
        }
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    f"{self.api_base}/api/execute",
                    json=payload,
                    headers=headers,
                    timeout=aiohttp.ClientTimeout(total=request.timeout)
                ) as response:
                    data = await response.json()
                    
                    duration_ms = int((asyncio.get_event_loop().time() - start_time) * 1000)
                    
                    if response.status == 200:
                        return AgentExecutionResponse(
                            success=True,
                            result=data.get("result"),
                            session_id=data.get("session_id"),
                            duration_ms=duration_ms,
                            metadata=data.get("metadata", {})
                        )
                    else:
                        return AgentExecutionResponse(
                            success=False,
                            error=data.get("error", "Unknown error"),
                            duration_ms=duration_ms
                        )
        except Exception as e:
            return AgentExecutionResponse(
                success=False,
                error=str(e),
                duration_ms=int((asyncio.get_event_loop().time() - start_time) * 1000)
            )
    
    async def health_check(self) -> bool:
        """检查OpenCAW服务状态"""
        import aiohttp
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    f"{self.api_base}/health",
                    timeout=aiohttp.ClientTimeout(total=5)
                ) as response:
                    return response.status == 200
        except Exception:
            return False
    
    async def get_status(self) -> dict:
        """获取OpenCAW状态"""
        import aiohttp
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    f"{self.api_base}/api/status",
                    headers={"Authorization": f"Bearer {self.api_key}"}
                ) as response:
                    return await response.json()
        except Exception as e:
            return {"error": str(e)}

class MCPBridge:
    """MCP协议桥接器 - 连接平台与外部Agent"""
    
    def __init__(self):
        self.servers: dict[str, MCPServer] = {}
        self.tools_cache: dict[str, list[ToolSchema]] = {}
    
    async def connect_server(
        self,
        name: str,
        command: str,
        args: list[str] = None,
        env: dict = None,
        transport: str = "stdio"  # stdio | http | sse
    ) -> bool:
        """连接MCP服务器"""
        server = MCPServer(
            name=name,
            command=command,
            args=args or [],
            env=env or {},
            transport=transport
        )
        
        try:
            await server.start()
            await server.initialize()
            
            # 获取可用工具
            tools = await server.list_tools()
            self.servers[name] = server
            self.tools_cache[name] = tools
            
            logger.info(f"Connected MCP server: {name} with {len(tools)} tools")
            return True
        except Exception as e:
            logger.error(f"Failed to connect MCP server {name}: {e}")
            return False
    
    async def disconnect_server(self, name: str):
        """断开MCP服务器"""
        if name in self.servers:
            await self.servers[name].stop()
            del self.servers[name]
            del self.tools_cache[name]
    
    async def call_tool(
        self,
        server_name: str,
        tool_name: str,
        arguments: dict
    ) -> dict:
        """调用MCP工具"""
        server = self.servers.get(server_name)
        if not server:
            raise MCPServerNotFoundError(f"Server not found: {server_name}")
        
        return await server.call_tool(tool_name, arguments)
    
    def get_all_tools(self) -> list[dict]:
        """获取所有MCP服务器的工具清单"""
        all_tools = []
        for server_name, tools in self.tools_cache.items():
            for tool in tools:
                all_tools.append({
                    "server": server_name,
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.input_schema
                })
        return all_tools


@dataclass
class MCPServer:
    name: str
    command: str
    args: list[str]
    env: dict
    transport: str
    
    process: asyncio.subprocess.Process | None = None
    capabilities: dict = field(default_factory=dict)
    
    async def start(self):
        """启动MCP服务器进程"""
        env = {**os.environ, **self.env}
        self.process = await asyncio.create_subprocess_exec(
            self.command, *self.args,
            env=env,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE
        )
    
    async def initialize(self):
        """初始化MCP协议"""
        # 发送initialize请求
        request = {
            "jsonrpc": "2.0",
            "id": 0,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "ai-platform", "version": "1.0"}
            }
        }
        # ...协议处理

class AgentFrameworkRegistry:
    """Agent框架注册中心 - 统一管理所有外部Agent"""
    
    def __init__(self):
        self._adapters: dict[AgentFramework, ExternalAgentAdapter] = {}
        self._metadata: dict[AgentFramework, AgentFrameworkMetadata] = {}
        self._mcp_bridge = MCPBridge()
        self._execution_history: list[AgentExecutionResponse] = []
    
    def register(
        self,
        framework: AgentFramework,
        adapter: ExternalAgentAdapter
    ):
        """注册Agent框架"""
        self._adapters[framework] = adapter
        self._metadata[framework] = adapter.get_metadata()
        logger.info(f"Registered agent framework: {framework.value}")
    
    def unregister(self, framework: AgentFramework):
        """注销Agent框架"""
        if framework in self._adapters:
            del self._adapters[framework]
            del self._metadata[framework]
            logger.info(f"Unregistered agent framework: {framework.value}")
    
    def get_adapter(self, framework: AgentFramework) -> ExternalAgentAdapter | None:
        """获取适配器"""
        return self._adapters.get(framework)
    
    def list_frameworks(self) -> list[AgentFrameworkMetadata]:
        """列出所有注册的框架"""
        return list(self._metadata.values())
    
    async def execute(
        self,
        request: AgentExecutionRequest
    ) -> AgentExecutionResponse:
        """统一执行入口"""
        adapter = self._adapters.get(request.framework)
        if not adapter:
            raise AgentFrameworkNotFoundError(
                f"Framework not supported: {request.framework.value}"
            )
        
        response = await adapter.execute(request)
        self._execution_history.append(response)
        
        # 记录执行统计
        await self._record_execution_stats(request.framework, response)
        
        return response
    
    async def execute_with_fallback(
        self,
        request: AgentExecutionRequest,
        fallback_frameworks: list[AgentFramework]
    ) -> AgentExecutionResponse:
        """带降级的执行 - 主框架失败时尝试备选框架"""
        # 首先尝试主框架
        try:
            response = await self.execute(request)
            if response.success:
                return response
        except Exception as e:
            logger.warning(f"Primary framework failed: {e}")
        
        # 尝试备选框架
        for framework in fallback_frameworks:
            try:
                request.framework = framework
                response = await self.execute(request)
                if response.success:
                    logger.info(f"Fallback to {framework.value} succeeded")
                    return response
            except Exception as e:
                logger.warning(f"Fallback framework {framework.value} failed: {e}")
                continue
        
        return AgentExecutionResponse(
            success=False,
            error="All frameworks failed"
        )
    
    async def health_check_all(self) -> dict[AgentFramework, bool]:
        """检查所有Agent框架健康状态"""
        results = {}
        for framework, adapter in self._adapters.items():
            try:
                results[framework] = await adapter.health_check()
            except Exception:
                results[framework] = False
        return results

┌─────────────────────────────────────────────────────────────────────────────┐
│                         外部Agent框架集成层 │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│    ┌─────────────────────────────────────────────────────────────────┐     │
│    │              Agent Framework Registry (注册中心)                 │     │
│    │                                                                 │     │
│    │   ┌────────────┐  ┌────────────┐  ┌────────────┐  ┌──────────┐ │     │
│    │   │Claude Code │  │  OpenCAW   │  │   Codex │  │  Custom  │ │     │
│    │   │  Adapter │  │  Adapter   │  │  Adapter   │  │  Adapter │ │     │
│    │   └────────────┘  └────────────┘  └────────────┘  └──────────┘ │     │
│    │ │              │              │               │        │     │
│    └──────────┼──────────────┼──────────────┼───────────────┼────────┘     │
│               │              │              │               │              │
│               ▼              ▼              ▼               ▼              │
│    ┌─────────────────────────────────────────────────────────────────┐     │
│    │                    MCP Bridge (协议桥接)                         │     │
│    │                                                                 │     │
│    │   ┌────────────┐  ┌────────────┐  ┌────────────┐  ┌──────────┐ │     │
│    │   │   MCP      │  │   MCP      │  │   MCP      │  │   MCP    │ │     │
│    │   │  Server 1  │  │  Server 2  │  │  Server 3  │  │   ... │ │     │
│    │   │ (GitHub)   │  │ ( Postgres) │ │ (Filesystem) │ │    │     │
│    │  └────────────┘  └────────────┘ └────────────┘  └──────────┘ │     │
│   └─────────────────────────────────────────────────────────────────┘     │
│ │                                            │
└──────────────────────────────┼──────────────────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         平台编排调度层                                      │
│             │
│   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐      │
│   │ Agent │  │  Workflow  │  │    Task     │  │   Event     │      │
│   │ Runtime     │  │   Engine   │  │  Scheduler  │  │    Bus      │      │
│   └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘      │
│                                                                             │
│   ┌─────────────────────────────────────────────────────────────────┐     │
│   │ Execution Router (执行路由)                         │     │
│   │   • 框架选择策略  • 降级策略  • 并发控制  • 超时管理 │     │
│   └─────────────────────────────────────────────────────────────────┘     │
└─────────────────────────────────────────────────────────────────────────────┘

# Step 1: 实现适配器
class NewFrameworkAdapter(ExternalAgentAdapter):
    def __init__(self, config: dict):
        super().__init__(config)
        # 初始化配置
    
    def get_metadata(self) -> AgentFrameworkMetadata:
        return AgentFrameworkMetadata(
            name="NewFramework",
            version="1.0",
            capabilities=["..."],
            supported_languages=["..."],
            requires_permissions=["..."],
            config_schema={},
            auth_method="..."
        )
    
    async def execute(self, request: AgentExecutionRequest) -> AgentExecutionResponse:
        # 实现执行逻辑
        pass
    
    async def health_check(self) -> bool:
        # 健康检查
        pass

# Step 2: 注册到框架注册中心
registry = AgentFrameworkRegistry()
registry.register(
    AgentFramework("new_framework"),
    NewFrameworkAdapter({"api_key": "..."})
)

# Step 3: 在平台中使用
response = await registry.execute(AgentExecutionRequest(
    task="帮我修复这个bug",
    framework=AgentFramework("new_framework"),
    working_dir="/path/to/project"
))

┌─────────────────────────────────────────────────────────────────────────────┐
│                                                                             │
│    ┌───────────────────────────────────────────────────────────────────┐   │
│    │                        数据资产 (Data Assets)                        │   │
│    │                                                                     │   │
│    │   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │   │
│    │   │   知识库    │  │  对话记录   │  │  使用痕迹   │              │   │
│    │   │ Knowledge  │  │  Sessions   │  │  Usage Logs │              │   │
│    │   └─────────────┘  └─────────────┘  └─────────────┘              │   │
│    │                                                                     │   │
│    │   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │   │
│    │   │   智能体    │  │  工作流     │  │   配置文件  │              │   │
│    │   │  Config    │  │  Config    │  │   Settings  │              │   │
│    │   └─────────────┘  └─────────────┘  └─────────────┘              │   │
│    │                                                                     │   │
│    │   ★ 持久化存储 ★ 跨版本迁移 ★ 数据主权 ★                         │   │
│    │                                                                     │   │
│    └───────────────────────────────────────────────────────────────────┘   │
│                                    │                                         │
│                                    ▼                                         │
│    ┌───────────────────────────────────────────────────────────────────┐   │
│    │                        Agent工具 (Agent Tools)                      │   │
│    │                                                                     │   │
│    │   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │   │
│    │   │ Claude Code │  │  OpenCAW   │  │   Custom    │              │   │
│    │   └─────────────┘  └─────────────┘  └─────────────┘              │   │
│    │                                                                     │   │
│    │   • 即插即用    • 版本迭代    • 随时替换                          │   │
│    │                                                                     │   │
│    └───────────────────────────────────────────────────────────────────┘   │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

# ============================================================================
# 用户数据层 (必须持久化)
# ============================================================================

@dataclass
class UserData:
    """用户核心数据 - 永久存储"""
    id: str
    name: str
    email: str
    tenant_id: str
    
    # 创建的数据
    created_agents: list[str]      # 创建的智能体ID列表
    created_workflows: list[str]    # 创建的工作流ID列表
    created_knowledge: list[str]   # 创建的知识库ID列表
    
    # 配置与偏好
    preferences: UserPreferences
    api_keys: list[APIKey]         # 绑定的API密钥
    
    # 元数据
    created_at: datetime
    updated_at: datetime

@dataclass
class UserPreferences:
    """用户偏好设置"""
    default_model: str
    theme: str = "light"
    language: str = "zh-CN"
    notification_enabled: bool = True

@dataclass
class AgentConfig:
    """智能体配置 - 持久化"""
    id: str
    user_id: str                    # 所有者
    
    # 配置内容（持久化的核心）
    name: str
    description: str
    instructions: str               # 系统提示词
    model_config: ModelConfig       # 模型配置
    
    # 关联数据
    tool_ids: list[str]            # 关联的工具ID
    knowledge_ids: list[str]       # 关联的知识库ID
    
    # 版本控制
    version: int
    version_history: list[AgentVersion]
    
    # 公开属性
    is_public: bool = False
    tags: list[str] = field(default_factory=list)
    
    created_at: datetime
    updated_at: datetime

@dataclass
class WorkflowConfig:
    """工作流配置 - 持久化"""
    id: str
    user_id: str
    project_id: str
    
    name: str
    description: str
    nodes: list[WorkflowNode]      # 节点定义（持久化）
    edges: list[WorkflowEdge]      # 连线定义（持久化）
    
    trigger_config: dict
    version: int
    
    is_public: bool = False
    
    created_at: datetime
    updated_at: datetime

# ============================================================================
# 知识库 - 持久化 (分层设计)
# ============================================================================

@dataclass
class KnowledgeBase:
    """知识库 - 分层持久化设计"""
    id: str
    user_id: str
    project_id: str
    
    name: str
    description: str
    
    # ========== 持久化层 (原始数据，不可丢失) ==========
    
    # 1. 原始文档 - 对象存储 (真正持久化)
    raw_documents: list[RawDocument]       # PDF/Word/MD等原始文件
    
    # 2. 解析后的结构化数据 - PostgreSQL
    parsed_content: list[ParsedDocument]   # 提取的文本、结构
    
    # 3. 元数据 - PostgreSQL
    metadata: KnowledgeMetadata
    
    # ========== 可重建索引层 (可以从原始数据重新生成) ==========
    
    # 4. 向量索引 - 向量数据库 (可重建)
    vector_index_id: str | None            # 指向向量数据库中的索引
    embedding_model: str
    
    # 5. 统计信息 - PostgreSQL
    doc_count: int = 0
    chunk_count: int = 0
    total_size_bytes: int = 0
    
    created_at: datetime
    updated_at: datetime

@dataclass
class RawDocument:
    """原始文档 - 真正持久化到对象存储"""
    id: str
    knowledge_base_id: str
    
    # 存储位置（对象存储路径）
    storage_path: str                       # s3://bucket/knowledge/{kb_id}/{doc_id}/file.pdf
    storage_bucket: str
    storage_region: str
    
    # 文件信息
    filename: str
    file_size: int
    mime_type: str                          # application/pdf, text/markdown, ...
    
    # 内容哈希（完整性校验）
    content_hash: str                       # SHA256
    
    # 版本控制
    version: int
    previous_versions: list[str]           # 历史版本ID列表
    
    uploaded_at: datetime
    updated_at: datetime

@dataclass
class ParsedDocument:
    """解析后的文档内容 - PostgreSQL持久化"""
    id: str
    raw_document_id: str
    
    # 解析后的文本内容（原始文本副本）
    content: str                            # 完整文本内容
    content_length: int
    
    # 结构化提取
    title: str | None
    author: str | None
    created_date: datetime | None
    sections: list[DocumentSection]        # 分节信息
    
    # 预处理标记
    language: str
    encoding: str
    
    parsed_at: datetime

@dataclass
class DocumentSection:
    """文档章节"""
    title: str
    start_char: int
    end_char: int
    level: int                              # 1=h1, 2=h2, 3=h3...
    subsections: list[DocumentSection] = field(default_factory=list)

@dataclass
class KnowledgeMetadata:
    """知识库元数据 - PostgreSQL"""
    id: str
    knowledge_base_id: str
    
    # 索引状态
    index_status: IndexStatus               # pending/processing/completed/failed
    vector_index_id: str | None            # 向量数据库中的索引ID
    
    # 嵌入配置
    embedding_model: str                   # "text-embedding-3-small"
    chunk_size: int = 512                   # 块大小
    chunk_overlap: int = 50                # 重叠大小
    
    # 统计
    total_documents: int = 0
    total_chunks: int = 0
    indexed_chunks: int = 0
    
    # 最后一次索引时间
    last_indexed_at: datetime | None
    last_error: str | None

@dataclass
class VectorIndex:
    """向量索引记录 - 记录索引元信息，可重建"""
    id: str
    knowledge_base_id: str
    
    # 索引来源
    source_document_ids: list[str]         # 关联的原始文档
    
    # 索引配置
    embedding_model: str
    dimension: int
    metric: str                             # "cosine" / "euclidean"
    
    # 向量数据库连接信息
    vector_db_type: str                    # "milvus" / "qdrant" / "pinecone"
    vector_db_endpoint: str
    collection_name: str
    
    # 状态
    status: str                            # "active" / "rebuilding" / "failed"
    
    # 统计
    total_vectors: int = 0
    vector_count_by_doc: dict[str, int] = field(default_factory=dict)  # doc_id -> count
    
    created_at: datetime
    updated_at: datetime