Agent Harness 技术原理与应用

┌─────────────────────────────────────────────────────────────┐
│                    控制平面 (Control Plane)                  │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  API Gateway│  │  认证授权   │  │  速率限制   │        │
│  │  (入口)     │  │  (Auth)     │  │  (Rate Limit)│       │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘        │
│         │                │                │                 │
│         └────────────────┴────────────────┘                 │
│                          │                                  │
│                          ▼                                  │
│  ┌─────────────────────────────────────────────────┐       │
│  │            任务调度器 (Task Scheduler)           │       │
│  │  • 任务队列管理                                 │       │
│  │  • 优先级调度                                   │       │
│  │  • 负载均衡                                     │       │
│  └──────────────────┬──────────────────────────────┘       │
│                     │                                       │
│         ┌───────────┼───────────┐                          │
│         │           │           │                          │
│         ▼           ▼           ▼                          │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐          │
│  │ 状态管理器  │ │ 策略引擎    │ │ 资源管理器  │          │
│  │ (State Mgr) │ │ (Policy)    │ │ (Resource)  │          │
│  │             │ │             │ │             │          │
│  │ • 状态追踪  │ │ • 安全策略  │ │ • 配额管理  │          │
│  │ • 持久化    │ │ • 权限控制  │ │ • 限流      │          │
│  │ • 查询      │ │ • 合规检查  │ │ • 优化      │          │
│  └──────┬──────┘ └──────┬──────┘ └──────┬──────┘          │
│         │               │               │                   │
│         └───────────────┴───────────────┘                   │
│                          │                                  │
│                          ▼                                  │
│  ┌─────────────────────────────────────────────────┐       │
│  │          监控与可观测性 (Monitoring)             │       │
│  │  • 指标收集  • 日志聚合  • 分布式追踪           │       │
│  └─────────────────────────────────────────────────┘       │
│                                                             │
│                          │                                  │
│                          ▼                                  │
│              下发指令到数据平面 (Data Plane)                │
│                                                             │
└─────────────────────────────────────────────────────────────┘

关键特性：
  • 集中决策：所有全局决策由控制平面统一做出
  • 状态持久化：关键状态持久化存储，支持故障恢复
  • 策略统一：安全、权限、限流等策略集中管理
  • 可观测性：完整监控、日志、追踪能力
                        

from enum import Enum
from dataclasses import dataclass
from typing import List, Optional
import heapq
import time

class Priority(Enum):
    CRITICAL = 0
    HIGH = 1
    NORMAL = 2
    LOW = 3

@dataclass
class Task:
    id: str
    priority: Priority
    created_at: float
    payload: dict
    deadline: Optional[float] = None
    
    def __lt__(self, other):
        # 优先级队列比较：优先级高 + 截止时间近
        if self.priority.value != other.priority.value:
            return self.priority.value < other.priority.value
        if self.deadline and other.deadline:
            return self.deadline < other.deadline
        return self.created_at < other.created_at

class TaskScheduler:
    def __init__(self):
        self.task_queue: List[Task] = []
        self.active_tasks = {}
        self.max_concurrent = 100
    
    def submit(self, task: Task):
        """提交任务到调度队列"""
        heapq.heappush(self.task_queue, task)
        self._schedule()
    
    def _schedule(self):
        """调度任务执行"""
        while len(self.active_tasks) < self.max_concurrent and self.task_queue:
            task = heapq.heappop(self.task_queue)
            
            # 检查是否过期
            if task.deadline and time.time() > task.deadline:
                continue  # 跳过过期任务
            
            # 执行任务
            self.active_tasks[task.id] = task
            self._execute_async(task)
    
    def _execute_async(self, task: Task):
        """异步执行任务"""
        import asyncio
        asyncio.create_task(self._run_task(task))
    
    async def _run_task(self, task: Task):
        """运行任务"""
        try:
            # 下发到数据平面执行
            result = await self._dispatch_to_data_plane(task)
            self._on_task_complete(task.id, result)
        except Exception as e:
            self._on_task_failed(task.id, e)
    
    def _dispatch_to_data_plane(self, task: Task):
        """下发到数据平面"""
        # 实际实现：通过 RPC/消息队列下发
        pass
    
    def _on_task_complete(self, task_id: str, result):
        """任务完成回调"""
        del self.active_tasks[task_id]
        self._schedule()  # 调度下一个任务
    
    def _on_task_failed(self, task_id: str, error):
        """任务失败回调"""
        del self.active_tasks[task_id]
        # 重试逻辑或错误处理
        self._schedule()

# 使用示例
scheduler = TaskScheduler()

# 提交关键任务
critical_task = Task(
    id="task-001",
    priority=Priority.CRITICAL,
    created_at=time.time(),
    payload={"type": "research", "query": "AI 趋势"},
    deadline=time.time() + 300  # 5 分钟截止
)
scheduler.submit(critical_task)
                        

import docker
import json
from typing import Dict, Any

class DockerSandbox:
    def __init__(self, image: str, timeout: int = 300):
        self.client = docker.from_env()
        self.image = image
        self.timeout = timeout
        self.container = None
    
    def create(self, resources: Dict[str, Any] = None):
        """创建沙箱容器"""
        if resources is None:
            resources = {}
        
        self.container = self.client.containers.run(
            self.image,
            command="tail -f /dev/null",  # 保持运行
            detach=True,
            remove=True,
            network_disabled=True,  # 默认禁用网络
            mem_limit=resources.get("memory", "512m"),
            cpu_quota=resources.get("cpu_quota", 50000),  # 50% CPU
            pids_limit=resources.get("pids", 100),
            read_only=True,  # 只读文件系统
            tmpfs={"/tmp": "rw,noexec,nosuid,size=100m"},  # 临时目录
            cap_drop=["ALL"],  # 删除所有能力
            security_opt=["no-new-privileges"],  # 禁止提权
        )
        return self.container.id
    
    def execute(self, command: str, timeout: int = None):
        """在沙箱中执行命令"""
        if not self.container:
            raise RuntimeError("沙箱未创建")
        
        result = self.container.exec_run(
            command,
            demux=True,
            workdir="/tmp",
            user="nobody",  # 非 root 用户
        )
        
        stdout, stderr = result.output
        return {
            "exit_code": result.exit_code,
            "stdout": stdout.decode() if stdout else "",
            "stderr": stderr.decode() if stderr else "",
        }
    
    def upload(self, content: str, path: str):
        """上传文件到沙箱"""
        import tarfile
        import io
        
        tar_stream = io.BytesIO()
        with tarfile.open(fileobj=tar_stream, mode='w') as tar:
            tarinfo = tarfile.TarInfo(name=path)
            tarinfo.size = len(content)
            tar.addfile(tarinfo, io.BytesIO(content.encode()))
        
        tar_stream.seek(0)
        self.container.put_archive("/tmp", tar_stream)
    
    def destroy(self):
        """销毁沙箱"""
        if self.container:
            self.container.stop(timeout=5)
            self.container = None

# 使用示例
sandbox = DockerSandbox(image="python:3.11-slim", timeout=300)
sandbox.create(resources={"memory": "256m", "cpu_quota": 25000})

# 上传代码
code = """
print("Hello from sandbox!")
result = 2 + 2
print(f"Result: {result}")
"""
sandbox.upload(code, "script.py")

# 执行代码
result = sandbox.execute("python /tmp/script.py", timeout=10)
print(result)

# 销毁沙箱
sandbox.destroy()
                        

from enum import Enum
from typing import Set, Dict

class Role(Enum):
    ADMIN = "admin"
    DEVELOPER = "developer"
    ANALYST = "analyst"
    VIEWER = "viewer"

class Permission(Enum):
    # 工具调用权限
    TOOL_SEARCH = "tool:search"
    TOOL_CODE_EXEC = "tool:code_exec"
    TOOL_FILE_READ = "tool:file_read"
    TOOL_FILE_WRITE = "tool:file_write"
    TOOL_NETWORK = "tool:network"
    
    # 管理权限
    AGENT_CREATE = "agent:create"
    AGENT_DELETE = "agent:delete"
    POLICY_MANAGE = "policy:manage"

# 角色 - 权限映射
ROLE_PERMISSIONS: Dict[Role, Set[Permission]] = {
    Role.ADMIN: set(Permission),  # 所有权限
    Role.DEVELOPER: {
        Permission.TOOL_SEARCH,
        Permission.TOOL_CODE_EXEC,
        Permission.TOOL_FILE_READ,
        Permission.TOOL_FILE_WRITE,
        Permission.AGENT_CREATE,
    },
    Role.ANALYST: {
        Permission.TOOL_SEARCH,
        Permission.TOOL_FILE_READ,
    },
    Role.VIEWER: set(),  # 只读权限
}

class AuthorizationEngine:
    def __init__(self):
        self.user_roles: Dict[str, Role] = {}
    
    def assign_role(self, user_id: str, role: Role):
        """分配角色给用户"""
        self.user_roles[user_id] = role
    
    def check_permission(self, user_id: str, permission: Permission) -> bool:
        """检查用户是否有权限"""
        role = self.user_roles.get(user_id)
        if not role:
            return False
        
        allowed = ROLE_PERMISSIONS.get(role, set())
        return permission in allowed
    
    def enforce(self, user_id: str, permission: Permission):
        """强制权限检查"""
        if not self.check_permission(user_id, permission):
            raise PermissionError(
                f"User {user_id} lacks permission {permission.value}"
            )

# 使用示例
auth_engine = AuthorizationEngine()
auth_engine.assign_role("user-001", Role.DEVELOPER)

# 检查权限
if auth_engine.check_permission("user-001", Permission.TOOL_CODE_EXEC):
    print("允许执行代码")
else:
    print("拒绝执行代码")

# 强制权限检查
try:
    auth_engine.enforce("user-001", Permission.TOOL_NETWORK)
except PermissionError as e:
    print(f"权限错误：{e}")