X-Agents/docs/ARCHITECTURE.md

X-Agents 智能体平台架构设计

一、整体架构

┌─────────────────────────────────────────────────────────────────────────────┐
│                                 用户层                                       │
│                      Web / App / API Consumer                                │
└─────────────────────────────────┬───────────────────────────────────────────┘
                                  │ HTTP / WebSocket
                                  ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                            Go API Gateway                                    │
│  ┌─────────────────────────────────────────────────────────────────────────┐│
│  │  • HTTP Server (Gin)          • 认证鉴权 (JWT)                        ││
│  │  • 路由管理                    • 限流熔断                              ││
│  │  • 业务逻辑                    • 日志监控                              ││
│  │  • 数据库操作                  • 权限控制 (RBAC)                      ││
│  └─────────────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────┬───────────────────────────────────────────┘
                                  │ HTTP JSON API
                                  ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                          Python Agent Engine                                 │
│  ┌─────────────────────────────────────────────────────────────────────────┐│
│  │  • FastAPI Server          • Agent Core (LangChain/AutoGen)          ││
│  │  • LLM Adapter             • Tool Registry (白名单)                   ││
│  │  • Memory Manager          • Sandbox Executor (沙盒)                 ││
│  │  • RAG Pipeline            • Audit Logger                             ││
│  └─────────────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────────────┘

---

二、系统分层

2.1 Go 后端层 (server/)

server/                          # Go API Gateway 服务
├── cmd/api/                    # 程序入口
│   └── main.go
├── internal/
│   ├── config/                 # 配置管理
│   │   └── config.go
│   ├── handler/                 # HTTP处理器
│   │   ├── auth_handler.go     # 认证接口
│   │   ├── chat_handler.go     # 聊天接口
│   │   └── approval_handler.go # 审批接口
│   ├── service/                # 业务逻辑
│   │   ├── auth_service.go
│   │   ├── chat_service.go
│   │   └── approval_service.go
│   ├── repository/              # 数据访问层
│   │   ├── user_repo.go
│   │   ├── agent_repo.go
│   │   └── audit_repo.go
│   ├── middleware/              # 中间件
│   │   └── auth.go
│   └── model/                   # 数据模型
│       ├── user.go
│       ├── agent.go
│       └── audit.go
├── config/                      # 配置文件
│   └── config.yaml
├── Dockerfile
├── go.mod
└── go.sum

2.2 Python Agent 层 (agent/)

agent/                          # Python Agent Engine
├── app/
│   ├── main.py                 # FastAPI入口
│   ├── api/
│   │   └── routes.py           # API路由
│   ├── agent/
│   │   ├── core/
│   │   │   ├── agent.py        # Agent管理器
│   │   │   └── executor.py    # Agent执行器
│   │   ├── tools/
│   │   │   ├── registry.py   # 工具注册表(白名单)
│   │   │   └── impl/         # 工具实现
│   │   │       ├── search.py
│   │   │       ├── calculator.py
│   │   │       └── time_tool.py
│   │   └── memory/
│   │       └── session.py     # 会话管理
│   ├── llm/
│   │   └── factory.py         # LLM工厂
│   └── security/
│       ├── audit.py           # 审计日志
│       └── approval.py        # 审批服务
├── requirements.txt
├── Dockerfile
└── pyproject.toml

2.3 根目录结构

X-Agents/
├── server/                      # Go API Gateway
├── agent/                       # Python Agent Engine
├── web/                        # 前端 (Vue.js)
├── docs/
│   └── ARCHITECTURE.md        # 架构文档
├── docker-compose.yml          # 容器编排
├── .env.example               # 环境变量模板
└── README.md
│   ├── service/                # 业务逻辑
│   │   ├── chat_service.go
│   │   ├── agent_service.go
│   │   └── approval_service.go  # 审批服务
│   ├── repository/              # 数据访问层
│   │   ├── user_repo.go
│   │   ├── agent_repo.go
│   │   └── audit_repo.go
│   ├── middleware/              # 中间件
│   │   ├── auth.go              # 认证中间件
│   │   ├── rbac.go              # 权限中间件
│   │   └── audit.go              # 审计中间件
│   ├── client/                 # 外部服务客户端
│   │   └── python_client.go     # Python服务HTTP客户端
│   └── model/                   # 数据模型
│       ├── user.go
│       ├── agent.go
│       └── audit.go
├── pkg/
│   ├── utils/                   # 工具函数
│   └── errors/                 # 错误定义
└── go.mod

2.2 Python AI 层 (智能逻辑)

python/
├── app/
│   ├── main.py                 # FastAPI入口
│   ├── api/
│   │   ├── routes.py           # 路由定义
│   │   └── dependencies.py     # 依赖注入
│   ├── agent/
│   │   ├── core/
│   │   │   ├── agent.py        # Agent核心
│   │   │   ├── executor.py     # 执行器
│   │   │   └── memory.py       # 记忆管理
│   │   ├── tools/
│   │   │   ├── registry.py    # 工具注册表
│   │   │   ├── base.py         # 工具基类
│   │   │   ├── security.py     # 安全检查
│   │   │   └── impl/           # 具体工具实现
│   │   │       ├── search.py
│   │   │       ├── calculator.py
│   │   │       ├── database.py
│   │   │       └── sandbox.py  # 沙盒执行
│   │   └── sandbox/
│   │       ├── docker_sandbox.py
│   │       └── wasm_sandbox.py
│   ├── llm/
│   │   ├── factory.py          # LLM工厂
│   │   ├── openai_adapter.py
│   │   ├── anthropic_adapter.py
│   │   └── base.py
│   ├── rag/
│   │   ├── vector_store.py     # 向量存储
│   │   ├── retriever.py        # 检索器
│   │   └── pipeline.py         # RAG流程
│   └── security/
│       ├── permission.py       # 权限检查
│       ├── approval.py         # 审批管理
│       └── audit.py             # 安全审计
├── requirements.txt
└── pyproject.toml

---

三、通信机制

3.1 HTTP API 通信

┌──────────────────┐    HTTP POST     ┌──────────────────┐
│                  │  ─────────────▶  │                  │
│   Go Service     │   JSON Request   │  Python Service  │
│   (Port: 8080)  │                  │  (Port: 8081)    │
│                  │  ◀─────────────  │                  │
└──────────────────┘   JSON Response  └──────────────────┘

接口设计

1. Agent 聊天接口

POST /api/v1/agent/chat
Content-Type: application/json
Authorization: Bearer <token>

Request:
{
    "agent_id": "agent_001",
    "message": "帮我查询用户数据",
    "session_id": "session_xxx",
    "context": {}  // 额外上下文
}

Response:
{
    "reply": "查询结果...",
    "session_id": "session_xxx",
    "tools_used": ["query_database"],
    "metadata": {}
}

2. 工具执行审批接口

POST /api/v1/tool/approve
Request:
{
    "request_id": "req_001",
    "tool_name": "execute_sql",
    "params": {"sql": "SELECT * FROM users"},
    "reason": "用户查询自己的订单",
    "approved": true  // true=批准, false=拒绝
}

3. 工具执行状态查询

GET /api/v1/tool/request/{request_id}
Response:
{
    "status": "pending|approved|rejected|completed",
    "tool_name": "execute_sql",
    "created_at": "2024-01-01T00:00:00Z",
    "result": null  // 如果已完成
}

3.2 Go → Python 客户端

// internal/client/python_client.go

package client

type PythonAgentClient struct {
    baseURL string
    client  *http.Client
}

type ChatRequest struct {
    AgentID   string                 `json:"agent_id"`
    Message    string                 `json:"message"`
    SessionID  string                 `json:"session_id"`
    Context    map[string]interface{} `json:"context"`
}

type ChatResponse struct {
    Reply      string                 `json:"reply"`
    SessionID  string                 `json:"session_id"`
    ToolsUsed  []string              `json:"tools_used"`
    Metadata   map[string]interface{} `json:"metadata"`
}

// Chat 调用Python Agent服务
func (c *PythonAgentClient) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) {
    // 1. 构建请求
    // 2. 添加超时
    // 3. 发送请求
    // 4. 处理响应
    // 5. 错误处理
}

---

四、沙盒安全机制

4.1 安全架构总览

┌─────────────────────────────────────────────────────────────────────────┐
│                            安全控制层                                     │
│                                                                          │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌──────────┐ │
│  │  权限管理    │    │  工具分级    │    │  人工审批    │    │  审计日志 │ │
│  │  (RBAC)     │    │  (白名单)    │    │  (Approval) │    │  (Audit) │ │
│  └──────┬──────┘    └──────┬──────┘    └──────┬──────┘    └────┬─────┘ │
└─────────┼─────────────────┼─────────────────┼─────────────────┼────────┘
          │                 │                 │                 │
          ▼                 ▼                 ▼                 ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                           Agent 执行层                                    │
│                                                                          │
│   User Request ─▶ Permission Check ─▶ Tool Lookup ─▶ Execute ─▶ Result │
│                            │                    │            │           │
│                            ▼                    ▼            ▼           │
│                     [Need Approval] ──▶ [Pending Queue] ──▶ [Notify]    │
└─────────────────────────────────────────────────────────────────────────┘

4.2 工具安全等级

# python/app/agent/tools/security.py

from enum import Enum
from dataclasses import dataclass
from typing import List, Callable

class SecurityLevel(Enum):
    """工具安全等级"""
    SAFE = "safe"           # 安全操作：搜索、计算、读取公开数据
    REVIEW = "review"      # 需要审核：修改数据、发送消息
    DANGER = "danger"      # 危险操作：删除数据、执行代码、敏感API

@dataclass
class ToolMetadata:
    """工具元数据"""
    name: str
    description: str
    security_level: SecurityLevel
    require_approval: bool          # 是否需要人工审批
    allowed_roles: List[str]         # 允许调用的角色
    rate_limit: int                 # 调用频率限制
    timeout: int                    # 超时时间(秒)

class ToolSecurity:
    """工具安全管理"""

    # 安全等级阈值
    APPROVAL_THRESHOLD = SecurityLevel.REVIEW

    @staticmethod
    def check_permission(tool: ToolMetadata, user_role: str) -> bool:
        """检查用户权限"""
        if user_role in tool.allowed_roles:
            return True
        return False

    @staticmethod
    def need_approval(tool: ToolMetadata) -> bool:
        """判断是否需要审批"""
        return tool.security_level.value >= ToolSecurity.APPROVAL_THRESHOLD.value

4.3 工具注册与执行

# python/app/agent/tools/registry.py

from typing import Dict, Callable, Any
from .security import ToolMetadata, SecurityLevel

class ToolRegistry:
    """工具注册表 - 白名单机制"""

    def __init__(self):
        self._tools: Dict[str, tuple[Callable, ToolMetadata]] = {}

    def register(
        self,
        name: str,
        func: Callable,
        security_level: SecurityLevel = SecurityLevel.SAFE,
        require_approval: bool = False,
        allowed_roles: List[str] = None,
        description: str = ""
    ):
        """注册工具到白名单"""
        metadata = ToolMetadata(
            name=name,
            description=description,
            security_level=security_level,
            require_approval=require_approval or security_level == SecurityLevel.REVIEW,
            allowed_roles=allowed_roles or ["user", "admin"],
            rate_limit=100,
            timeout=30
        )
        self._tools[name] = (func, metadata)

    def get_tool(self, name: str) -> tuple[Callable, ToolMetadata]:
        """获取工具（必须在白名单中）"""
        if name not in self._tools:
            raise ValueError(f"Tool '{name}' not found in whitelist")
        return self._tools[name]

    def list_tools(self) -> List[ToolMetadata]:
        """列出所有可用工具"""
        return [meta for _, meta in self._tools.values()]

4.4 沙盒执行

# python/app/agent/tools/sandbox/docker_sandbox.py

import subprocess
import tempfile
import shutil
import os
from typing import Any, Dict

class DockerSandbox:
    """Docker沙盒执行环境"""

    def __init__(self, image: str = "python-sandbox:latest", timeout: int = 30):
        self.image = image
        self.timeout = timeout

    def execute(self, code: str, language: str = "python") -> Dict[str, Any]:
        """在沙盒中执行代码"""

        # 1. 创建临时文件
        with tempfile.NamedTemporaryFile(
            mode='w',
            suffix=f'.{language}',
            delete=False
        ) as f:
            f.write(code)
            temp_path = f.name

        try:
            # 2. Docker容器执行
            result = subprocess.run(
                [
                    "docker", "run",
                    "--rm",
                    "--network", "none",           # 断网
                    "--memory", "256m",            # 内存限制
                    "--cpus", "0.5",               # CPU限制
                    "-v", f"{temp_path}:/code/{os.path.basename(temp_path)}",
                    self.image,
                    "python", f"/code/{os.path.basename(temp_path)}"
                ],
                capture_output=True,
                timeout=self.timeout
            )

            return {
                "success": result.returncode == 0,
                "output": result.stdout.decode(),
                "error": result.stderr.decode()
            }

        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "output": "",
                "error": "Execution timeout"
            }
        finally:
            # 3. 清理临时文件
            os.unlink(temp_path)

# 使用示例
@sandbox.execute
def execute_code(code: str) -> str:
    """安全执行用户代码"""
    pass

4.5 Human in the Loop (人工审批)

# python/app/security/approval.py

from enum import Enum
from dataclasses import dataclass
from typing import Optional
from datetime import datetime
import asyncio

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"

@dataclass
class ApprovalRequest:
    """审批请求"""
    request_id: str
    tool_name: str
    params: dict
    user_id: str
    reason: str
    status: ApprovalStatus
    created_at: datetime
    reviewed_at: Optional[datetime]
    reviewed_by: Optional[str]

class ApprovalService:
    """审批服务"""

    def __init__(self, http_client):
        self.client = http_client
        self.pending: Dict[str, ApprovalRequest] = {}

    async def request_approval(
        self,
        tool_name: str,
        params: dict,
        user_id: str,
        reason: str
    ) -> str:
        """请求审批"""
        request_id = generate_uuid()

        approval_req = ApprovalRequest(
            request_id=request_id,
            tool_name=tool_name,
            params=params,
            user_id=user_id,
            reason=reason,
            status=ApprovalStatus.PENDING,
            created_at=datetime.now(),
            reviewed_at=None,
            reviewed_by=None
        )

        self.pending[request_id] = approval_req

        # 通知Go后端有新审批
        await self.notify_go_service(approval_req)

        return request_id

    async def wait_for_approval(self, request_id: str, timeout: int = 300) -> bool:
        """等待审批结果"""
        start = datetime.now()

        while (datetime.now() - start).seconds < timeout:
            if request_id in self.pending:
                status = self.pending[request_id].status
                if status == ApprovalStatus.APPROVED:
                    return True
                elif status == ApprovalStatus.REJECTED:
                    return False
            await asyncio.sleep(1)

        raise TimeoutError("Approval request timeout")

4.6 全链路审计

# python/app/security/audit.py

from datetime import datetime
from typing import Any, Dict
import json

class AuditLogger:
    """审计日志"""

    def __init__(self, log_file: str = "audit.log"):
        self.log_file = log_file

    def log(
        self,
        action: str,
        user_id: str,
        agent_id: str,
        details: Dict[str, Any],
        result: str = "success"
    ):
        """记录审计日志"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "action": action,
            "user_id": user_id,
            "agent_id": agent_id,
            "details": details,
            "result": result
        }

        # 写入日志文件
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(entry) + '\n')

        # 发送到Go后端
        self.send_to_backend(entry)

    def log_tool_execution(
        self,
        user_id: str,
        tool_name: str,
        params: Dict[str, Any],
        approved: bool,
        result: Any
    ):
        """记录工具执行"""
        self.log(
            action="tool_execution",
            user_id=user_id,
            agent_id="",
            details={
                "tool_name": tool_name,
                "params": params,
                "approved": approved,
                "result_preview": str(result)[:100]
            }
        )

---

五、权限模型 (Go端)

5.1 用户角色

// go/internal/model/user.go

package model

// 权限级别
type PermissionLevel int

const (
    PermissionRead    PermissionLevel = 1  // 只读
    PermissionWrite   PermissionLevel = 2  // 读写
    PermissionExecute PermissionLevel = 3  // 可执行工具
    PermissionAdmin   PermissionLevel = 4  // 管理员
)

// 角色定义
type Role struct {
    ID          string           `json:"id"`
    Name        string           `json:"name"`
    Permissions []PermissionLevel `json:"permissions"`
}

// 用户
type User struct {
    ID       string   `json:"id"`
    Username string   `json:"username"`
    RoleID   string   `json:"role_id"`
    Role     *Role    `json:"role,omitempty"`
}

5.2 Agent定义

// go/internal/model/agent.go

package model

type Agent struct {
    ID          string            `json:"id"`
    Name        string            `json:"name"`
    Description string            `json:"description"`
    OwnerID     string            `json:"owner_id"`

    // Agent能力配置
    Capabilities []string         `json:"capabilities"`  // 可用工具列表
    MemoryLimit int64             `json:"memory_limit"` // 内存限制
    Timeout     int               `json:"timeout"`      // 超时时间

    // 安全配置
    SecurityLevel SecurityLevel   `json:"security_level"`
    AllowDangerousTools bool      `json:"allow_dangerous_tools"`

    CreatedAt   time.Time         `json:"created_at"`
    UpdatedAt   time.Time         `json:"updated_at"`
}

---

六、部署架构

6.1 Docker Compose

# docker-compose.yml

version: '3.8'

services:
  # Go API 服务
  go-api:
    build: ./go
    ports:
      - "8080:8080"
    environment:
      - DATABASE_URL=postgres://user:pass@db:5432/agents
      - PYTHON_SERVICE_URL=http://python-agent:8081
      - JWT_SECRET=your-secret
    depends_on:
      - db
      - python-agent

  # Python Agent 服务
  python-agent:
    build: ./python
    ports:
      - "8081:8081"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    volumes:
      - ./python/app:/app
      - /var/run/docker.sock:/var/run/docker.sock  # 如果需要Docker沙盒

  # 数据库
  db:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: agents
    volumes:
      - db-data:/var/lib/postgresql/data

  # Redis (缓存/会话)
  redis:
    image: redis:7
    volumes:
      - redis-data:/data

  # 向量数据库 (可选)
  qdrant:
    image: qdrant/qdrant
    volumes:
      - qdrant-data:/qdrant/storage

volumes:
  db-data:
  redis-data:
  qdrant-data:

---

七、开发流程

7.1 请求流程图

┌─────────┐     ┌─────────┐     ┌─────────┐     ┌─────────┐     ┌─────────┐
│  用户   │────▶│  Go    │────▶│ Python  │────▶│  LLM   │────▶│ 返回   │
│  请求   │     │  鉴权   │     │ Agent   │     │ +Tools │     │ 结果   │
└─────────┘     └─────────┘     └─────────┘     └─────────┘     └─────────┘
                    │               │
                    │               │
                    ▼               ▼
              ┌─────────┐     ┌─────────┐
              │ 检查    │     │ 权限    │
              │ 权限    │     │ 检查    │
              └─────────┘     └─────────┘
                                │
                                ▼
                    ┌─────────────────────┐
                    │ 工具安全等级判断     │
                    └─────────────────────┘
                                │
              ┌─────────────────┼─────────────────┐
              │                 │                 │
              ▼                 ▼                 ▼
        ┌──────────┐     ┌──────────┐     ┌──────────┐
        │ Safe     │     │ Review   │     │ Danger   │
        │ 直接执行  │     │ 等待审批  │     │ 拒绝执行 │
        └──────────┘     └──────────┘     └──────────┘

7.2 目录结构总览

X-Agents/
├── docs/
│   └── ARCHITECTURE.md          # 本文档
│
├── go/                          # Go 后端
│   ├── cmd/
│   ├── internal/
│   ├── pkg/
│   ├── go.mod
│   └── Dockerfile
│
├── python/                      # Python AI 层
│   ├── app/
│   │   ├── api/
│   │   ├── agent/
│   │   ├── llm/
│   │   ├── rag/
│   │   └── security/
│   ├── requirements.txt
│   └── Dockerfile
│
├── web/                         # 前端 (Vue)
│   ├── src/
│   └── package.json
│
├── docker-compose.yml           # 容器编排
└── README.md

---

八、总结

架构核心原则

原则	实现方式
分层治理	Go负责业务/权限，Python负责AI逻辑
安全优先	工具分级+权限控制+人工审批+审计日志
通信简洁	HTTP JSON API，后续可升级gRPC
可扩展	模块化设计，支持多Agent/多Python服务
可观测	全链路日志+监控

安全特性

• 工具白名单机制
• 安全等级分级 (Safe/Review/Danger)
• RBAC权限控制
• Human in the Loop 人工审批
• 沙盒执行环境 (Docker)
• 全链路审计日志

---

本文档将随项目开发持续更新