虚拟化技术在智能体工作流中的延迟与吞吐权衡及调优

agent-virt-tuning/
├── requirements.txt
├── config.yaml
├── scheduler/
│   ├── __init__.py
│   └── dynamic_scheduler.py
├── virtualization/
│   ├── __init__.py
│   ├── base.py
│   └── docker_env.py
├── workflow/
│   ├── __init__.py
│   ├── engine.py
│   └── skills.py
├── monitor/
│   ├── __init__.py
│   └── metrics.py
├── main.py
├── run_workload.py
└── tests/
    ├── __init__.py
    ├── test_scheduler.py
    └── test_workflow.py

docker>=6.0.0
pyyaml>=5.4
prometheus-client>=0.17.0
numpy>=1.21.0
pytest>=7.0.0

virtualization:
  type: "docker"  # 未来可扩展为 'gvisor', 'kata'
  docker:
    base_image: "python:3.9-slim"
    network: "agent-network"

workflow:
  max_queue_size: 100
  default_timeout_seconds: 30

scheduler:
  update_interval_seconds: 10
  latency_slo_milliseconds: 500  # 延迟服务水平目标
  strategy: "balanced"  # 可选: latency_first, throughput_first, balanced
  resource_adjustment:
    cpu_step: 0.2  # 每次调整CPU份额的步长（核心数）
    memory_step_mb: 128  # 每次调整内存的步长（MB）
    min_cpu: 0.5
    max_cpu: 4.0
    min_memory_mb: 256
    max_memory_mb: 2048

monitoring:
  prometheus_port: 8000

"""
虚拟化环境抽象基类定义。
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from dataclasses import dataclass

@dataclass
class ResourceSpec:
    """资源规格定义"""
    cpu_limit: float  # CPU核心数
    memory_limit_mb: int  # 内存限制（MB）

class ExecutionUnit(ABC):
    """执行单元抽象类，代表一个隔离的运行环境（如容器）"""

    def __init__(self, unit_id: str, resource_spec: ResourceSpec):
        self.unit_id = unit_id
        self.resource_spec = resource_spec
        self.status = "created"  # created, running, stopped, error

    @abstractmethod
    async def start(self):
        """启动执行单元"""
        pass

    @abstractmethod
    async def execute(self, command: str, input_data: str = "") -> Dict[str, Any]:
        """
        在执行单元内执行命令或调用。
        返回字典，至少包含 'success', 'output', 'error', 'execution_time_ms' 键。
        """
        pass

    @abstractmethod
    async def update_resources(self, new_spec: ResourceSpec) -> bool:
        """动态更新资源限制"""
        pass

    @abstractmethod
    async def stop(self):
        """停止并清理执行单元"""
        pass

class VirtualizationEnv(ABC):
    """虚拟化环境管理器抽象类"""

    @abstractmethod
    async def create_unit(self, unit_id: str, resource_spec: ResourceSpec, **kwargs) -> ExecutionUnit:
        """创建一个新的执行单元"""
        pass

    @abstractmethod
    async def list_units(self) -> Dict[str, ExecutionUnit]:
        """获取所有执行单元"""
        pass

"""
基于Docker的虚拟化环境实现。
"""
import docker
import asyncio
from typing import Dict, Any
from .base import VirtualizationEnv, ExecutionUnit, ResourceSpec
from concurrent.futures import ThreadPoolExecutor

class DockerExecutionUnit(ExecutionUnit):
    """Docker容器执行单元"""

    def __init__(self, unit_id: str, resource_spec: ResourceSpec, container):
        super().__init__(unit_id, resource_spec)
        self._container = container
        self._docker_client = docker.from_env()
        self._thread_pool = ThreadPoolExecutor(max_workers=2)

    async def start(self):
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(self._thread_pool, self._container.start)
        self.status = "running"

    async def execute(self, command: str, input_data: str = "") -> Dict[str, Any]:
        import time
        start_time = time.time()
        result = {"success": False, "output": "", "error": "", "execution_time_ms": 0}

        try:
            # 在容器内执行命令。这是一个简化示例，实际中可能需要更复杂的交互。
            exec_id = await asyncio.get_event_loop().run_in_executor(
                self._thread_pool,
                lambda: self._docker_client.api.exec_create(
                    self._container.id,
                    cmd=['sh', '-c', f'echo "{input_data[:100]}" | {command}'],  # 简单管道模拟输入
                    stdout=True,
                    stderr=True
                )
            )
            exec_output = await asyncio.get_event_loop().run_in_executor(
                self._thread_pool,
                lambda: self._docker_client.api.exec_start(exec_id['Id'], stream=False)
            )
            exec_inspect = await asyncio.get_event_loop().run_in_executor(
                self._thread_pool,
                lambda: self._docker_client.api.exec_inspect(exec_id['Id'])
            )

            result['output'] = exec_output.decode('utf-8') if isinstance(exec_output, bytes) else exec_output
            result['success'] = (exec_inspect['ExitCode'] == 0)
            if not result['success']:
                result['error'] = f"Exit code: {exec_inspect['ExitCode']}"
        except Exception as e:
            result['error'] = str(e)

        result['execution_time_ms'] = int((time.time() - start_time) * 1000)
        return result

    async def update_resources(self, new_spec: ResourceSpec) -> bool:
        try:
            # Docker容器资源更新（需要容器在运行状态）
            update_config = docker.types.Resources(
                cpu_quota=int(new_spec.cpu_limit * 100000),  # 将核心数转换为微秒配额
                cpu_period=100000,
                mem_limit=f"{new_spec.memory_limit_mb}m"
            )
            await asyncio.get_event_loop().run_in_executor(
                self._thread_pool,
                lambda: self._container.update(**update_config)
            )
            self.resource_spec = new_spec
            return True
        except Exception:
            return False

    async def stop(self):
        try:
            await asyncio.get_event_loop().run_in_executor(
                self._thread_pool,
                lambda: self._container.stop(timeout=5)
            )
        except Exception:
            pass
        try:
            await asyncio.get_event_loop().run_in_executor(
                self._thread_pool,
                lambda: self._container.remove()
            )
        except Exception:
            pass
        self.status = "stopped"
        self._thread_pool.shutdown(wait=False)

class DockerVirtualizationEnv(VirtualizationEnv):
    """Docker环境管理器"""

    def __init__(self, config: Dict[str, Any]):
        self._client = docker.from_env()
        self._config = config
        self._units: Dict[str, DockerExecutionUnit] = {}
        # 确保网络存在
        self._ensure_network()

    def _ensure_network(self):
        network_name = self._config['docker']['network']
        try:
            self._client.networks.get(network_name)
        except docker.errors.NotFound:
            self._client.networks.create(network_name, driver="bridge")

    async def create_unit(self, unit_id: str, resource_spec: ResourceSpec, **kwargs) -> DockerExecutionUnit:
        # 构建容器创建配置
        container_config = {
            'image': self._config['docker']['base_image'],
            'command': 'tail -f /dev/null',  # 保持容器运行
            'detach': True,
            'name': f'agent-unit-{unit_id}',
            'network': self._config['docker']['network'],
            'cpu_quota': int(resource_spec.cpu_limit * 100000),
            'cpu_period': 100000,
            'mem_limit': f'{resource_spec.memory_limit_mb}m',
            'environment': ['PYTHONUNBUFFERED=1'],
        }
        container = await asyncio.get_event_loop().run_in_executor(
            None, lambda: self._client.containers.create(**container_config)
        )
        unit = DockerExecutionUnit(unit_id, resource_spec, container)
        self._units[unit_id] = unit
        return unit

    async def list_units(self) -> Dict[str, DockerExecutionUnit]:
        return self._units.copy()

"""
定义智能体工作流中可用的技能（Skill）。
每个技能映射到执行单元中的一个命令或调用。
"""
from typing import Dict, Any

class Skill:
    def __init__(self, name: str, command: str, estimated_resource: Dict[str, Any]):
        self.name = name
        self.command = command
        self.estimated_resource = estimated_resource  # 预估资源消耗，e.g., {'cpu': 0.5, 'memory_mb': 200}

# 预定义的技能库
SKILL_REGISTRY = {
    "text_processor": Skill(
        name="text_processor",
        command="python -c \"import sys; data=sys.stdin.read().strip(); print(f'Processed: {data.upper()} [{len(data)} chars]')\"",
        estimated_resource={'cpu': 0.3, 'memory_mb': 150}
    ),
    "ai_inference": Skill(
        name="ai_inference",
        # 模拟一个轻量级AI推理，实际中会调用模型服务
        command="python -c \"import time, sys, random; data=sys.stdin.read(); time.sleep(random.uniform(0.05, 0.2)); print(f'AI Result for <<{data[:20]}...>>: score={random.random():.2f}')\"",
        estimated_resource={'cpu': 1.5, 'memory_mb': 800}
    ),
    "data_aggregator": Skill(
        name="data_aggregator",
        command="python -c \"import sys, json; inputs = [line.strip() for line in sys.stdin if line]; print(f'Aggregated {len(inputs)} items: {inputs[-1][:50]}...')\"",
        estimated_resource={'cpu': 0.2, 'memory_mb': 100}
    )
}

"""
智能体工作流引擎。
负责编排技能在执行单元中的执行顺序。
"""
import asyncio
import uuid
from typing import List, Dict, Any, Optional
from datetime import datetime
from ..virtualization.base import VirtualizationEnv, ExecutionUnit, ResourceSpec
from .skills import SKILL_REGISTRY, Skill

class WorkflowStep:
    def __init__(self, skill_name: str, input_data: str = ""):
        self.skill_name = skill_name
        self.input_data = input_data
        self.skill: Optional[Skill] = SKILL_REGISTRY.get(skill_name)
        self.result: Optional[Dict[str, Any]] = None
        self.execution_unit: Optional[ExecutionUnit] = None

class WorkflowInstance:
    """一个工作流执行实例"""
    def __init__(self, wf_id: str, steps: List[WorkflowStep]):
        self.id = wf_id
        self.steps = steps
        self.status = "pending"  # pending, running, completed, failed
        self.start_time: Optional[datetime] = None
        self.end_time: Optional[datetime] = None
        self.final_output: Optional[str] = None

class WorkflowEngine:
    def __init__(self, virt_env: VirtualizationEnv, max_queue_size: int = 100):
        self.virt_env = virt_env
        self.max_queue_size = max_queue_size
        self.pending_queue: asyncio.Queue[WorkflowInstance] = asyncio.Queue(maxsize=max_queue_size)
        self.active_instances: Dict[str, WorkflowInstance] = {}
        self._unit_skill_map: Dict[str, str] = {}  # unit_id -> skill_name
        self._skill_unit_pool: Dict[str, List[ExecutionUnit]] = {}  # skill_name -> list of units
        self._lock = asyncio.Lock()

    async def _ensure_unit_for_skill(self, skill_name: str, resource_spec: ResourceSpec) -> ExecutionUnit:
        """确保某个技能有可用的执行单元，如果没有则创建"""
        async with self._lock:
            if skill_name not in self._skill_unit_pool:
                self._skill_unit_pool[skill_name] = []
            pool = self._skill_unit_pool[skill_name]

            for unit in pool:
                if unit.status == "running":
                    return unit

            # 没有空闲单元，创建新的
            unit_id = f"{skill_name}-{uuid.uuid4().hex[:8]}"
            unit = await self.virt_env.create_unit(unit_id, resource_spec)
            await unit.start()
            self._unit_skill_map[unit.unit_id] = skill_name
            pool.append(unit)
            return unit

    async def submit_workflow(self, steps_def: List[Dict[str, Any]]) -> str:
        """提交一个工作流，返回工作流ID"""
        if self.pending_queue.full():
            raise RuntimeError("Workflow queue is full")

        steps = [WorkflowStep(step_def['skill'], step_def.get('input', '')) for step_def in steps_def]
        wf_id = f"wf-{uuid.uuid4().hex[:8]}"
        instance = WorkflowInstance(wf_id, steps)
        await self.pending_queue.put(instance)
        self.active_instances[wf_id] = instance
        return wf_id

    async def _execute_step(self, step: WorkflowStep) -> bool:
        """执行单个步骤"""
        if not step.skill:
            step.result = {'success': False, 'error': f"Skill '{step.skill_name}' not found"}
            return False

        # 根据技能预估的资源需求创建或获取执行单元
        est = step.skill.estimated_resource
        resource_spec = ResourceSpec(
            cpu_limit=est['cpu'],
            memory_limit_mb=est['memory_mb']
        )
        try:
            unit = await self._ensure_unit_for_skill(step.skill_name, resource_spec)
            step.execution_unit = unit
            step.result = await unit.execute(step.skill.command, step.input_data)
            return step.result['success']
        except Exception as e:
            step.result = {'success': False, 'error': str(e)}
            return False

    async def process_workflow(self, instance: WorkflowInstance):
        """处理一个工作流实例"""
        instance.status = "running"
        instance.start_time = datetime.utcnow()
        step_outputs = []

        for step in instance.steps:
            success = await self._execute_step(step)
            step_outputs.append(step.result.get('output', '') if success else f"Error: {step.result.get('error')}")
            if not success:
                instance.status = "failed"
                break

        if instance.status == "running":
            instance.status = "completed"
            instance.final_output = " | ".join(step_outputs)
        instance.end_time = datetime.utcnow()
        # 从活跃实例中移除
        self.active_instances.pop(instance.id, None)

    async def run(self):
        """启动工作流引擎处理循环"""
        print("Workflow engine started.")
        while True:
            try:
                instance = await self.pending_queue.get()
                asyncio.create_task(self.process_workflow(instance))
            except asyncio.CancelledError:
                break
            except Exception as e:
                print(f"Error processing workflow from queue: {e}")

"""
动态资源调度器。
根据监控指标调整执行单元的资源限制。
"""
import asyncio
import time
from typing import Dict, Any, List
from dataclasses import dataclass
from ..virtualization.base import VirtualizationEnv, ResourceSpec
from ..workflow.engine import WorkflowEngine
from ..monitor.metrics import workflow_latency_histogram, workflow_throughput_counter, resource_usage_gauge

@dataclass
class SchedulerMetrics:
    avg_latency_ms: float
    throughput_per_min: float
    queue_length: int
    unit_resources: Dict[str, ResourceSpec]  # unit_id -> current resource

class DynamicResourceScheduler:
    def __init__(self, virt_env: VirtualizationEnv, workflow_engine: WorkflowEngine, config: Dict[str, Any]):
        self.virt_env = virt_env
        self.engine = workflow_engine
        self.config = config['scheduler']
        self.metrics_history: List[SchedulerMetrics] = []
        self._is_running = False

    async def collect_metrics(self) -> SchedulerMetrics:
        """收集当前系统指标"""
        # 注意：这是一个简化的实现。在实际系统中，这些指标应从监控模块实时获取。
        # 这里我们模拟计算。
        avg_latency = 500.0  # 默认值，应从 histogram 计算
        throughput = 10.0    # 默认值，应从 counter 计算
        queue_length = self.engine.pending_queue.qsize()

        units = await self.virt_env.list_units()
        unit_resources = {uid: unit.resource_spec for uid, unit in units.items()}

        # 更新Prometheus指标（模拟）
        workflow_latency_histogram.labels(strategy=self.config['strategy']).observe(avg_latency / 1000.0) # 转换为秒
        workflow_throughput_counter.labels(strategy=self.config['strategy']).inc(int(throughput))
        for uid, spec in unit_resources.items():
            resource_usage_gauge.labels(unit_id=uid, resource_type='cpu').set(spec.cpu_limit)
            resource_usage_gauge.labels(unit_id=uid, resource_type='memory_mb').set(spec.memory_limit_mb)

        return SchedulerMetrics(avg_latency, throughput, queue_length, unit_resources)

    def _make_decision(self, metrics: SchedulerMetrics) -> Dict[str, ResourceSpec]:
        """基于指标和策略做出资源调整决策"""
        decisions = {}
        strategy = self.config['strategy']
        latency_slo = self.config['latency_slo_milliseconds']
        adj = self.config['resource_adjustment']

        # 简化决策逻辑：
        # 1. 如果平均延迟 > SLO，倾向于增加资源（尤其是高CPU消耗的技能）。
        # 2. 如果队列过长，增加资源以提升吞吐。
        # 3. 根据策略调整激进程度。
        scale_factor = 1.0
        if strategy == "latency_first":
            scale_factor = 1.5  # 更积极地增加资源以降低延迟
        elif strategy == "throughput_first":
            scale_factor = 0.7  # 更保守，容忍更高延迟以节省资源

        for unit_id, current_spec in metrics.unit_resources.items():
            new_cpu = current_spec.cpu_limit
            new_mem = current_spec.memory_limit_mb

            # 决策示例：如果延迟超标，增加CPU
            if metrics.avg_latency_ms > latency_slo:
                new_cpu = min(current_spec.cpu_limit + adj['cpu_step'] * scale_factor, adj['max_cpu'])
            elif metrics.avg_latency_ms < latency_slo * 0.7 and metrics.queue_length < 5:
                # 延迟很低且队列空闲，可以适当缩减资源
                new_cpu = max(current_spec.cpu_limit - adj['cpu_step'] * 0.5, adj['min_cpu'])

            # 内存调整逻辑（示例）：根据CPU比例调整
            cpu_ratio = new_cpu / current_spec.cpu_limit if current_spec.cpu_limit > 0 else 1.0
            new_mem = int(min(max(current_spec.memory_limit_mb * cpu_ratio, adj['min_memory_mb']), adj['max_memory_mb']))

            if abs(new_cpu - current_spec.cpu_limit) > 0.05 or abs(new_mem - current_spec.memory_limit_mb) > 10:
                decisions[unit_id] = ResourceSpec(cpu_limit=round(new_cpu, 2), memory_limit_mb=new_mem)

        return decisions

    async def _apply_decisions(self, decisions: Dict[str, ResourceSpec]):
        """应用资源调整决策到对应的执行单元"""
        units = await self.virt_env.list_units()
        for unit_id, new_spec in decisions.items():
            if unit_id in units:
                unit = units[unit_id]
                success = await unit.update_resources(new_spec)
                if success:
                    print(f"Scheduler: Updated {unit_id} resources to CPU={new_spec.cpu_limit}, Mem={new_spec.memory_limit_mb}MB")

    async def run_scheduling_loop(self):
        """调度器主循环"""
        self._is_running = True
        interval = self.config['update_interval_seconds']
        print(f"Dynamic scheduler started with strategy '{self.config['strategy']}', interval {interval}s.")

        while self._is_running:
            try:
                await asyncio.sleep(interval)
                metrics = await self.collect_metrics()
                self.metrics_history.append(metrics)
                # 保留最近100条记录
                if len(self.metrics_history) > 100:
                    self.metrics_history.pop(0)

                decisions = self._make_decision(metrics)
                if decisions:
                    await self._apply_decisions(decisions)

            except asyncio.CancelledError:
                break
            except Exception as e:
                print(f"Scheduler loop error: {e}")

    def stop(self):
        self._is_running = False

"""
Prometheus指标定义与暴露。
"""
from prometheus_client import start_http_server, Histogram, Counter, Gauge
import yaml

# 加载配置
with open('config.yaml', 'r') as f:
    config = yaml.safe_load(f)
prom_port = config['monitoring']['prometheus_port']

# 定义指标
workflow_latency_histogram = Histogram(
    'workflow_latency_seconds',
    'Latency of completed workflows in seconds',
    ['strategy']
)

workflow_throughput_counter = Counter(
    'workflow_throughput_total',
    'Total number of completed workflows',
    ['strategy']
)

resource_usage_gauge = Gauge(
    'execution_unit_resource',
    'Current resource allocation of an execution unit',
    ['unit_id', 'resource_type']  # resource_type: 'cpu' or 'memory_mb'
)

def start_monitoring_server():
    """启动Prometheus指标HTTP服务器"""
    start_http_server(prom_port)
    print(f"Monitoring server started on port {prom_port}")

"""
项目主入口，初始化并启动所有组件。
"""
import asyncio
import signal
import yaml
from workflow.engine import WorkflowEngine
from virtualization.docker_env import DockerVirtualizationEnv
from scheduler.dynamic_scheduler import DynamicResourceScheduler
from monitor.metrics import start_monitoring_server
import threading

def load_config():
    with open('config.yaml', 'r') as f:
        return yaml.safe_load(f)

async def main():
    config = load_config()
    print("Starting Agent Virtualization Tuning Demo...")

    # 1. 启动监控服务器（在独立线程中）
    threading.Thread(target=start_monitoring_server, daemon=True).start()

    # 2. 初始化虚拟化环境
    virt_env = DockerVirtualizationEnv(config['virtualization'])

    # 3. 初始化工作流引擎
    engine = WorkflowEngine(virt_env, config['workflow']['max_queue_size'])

    # 4. 初始化动态调度器
    scheduler = DynamicResourceScheduler(virt_env, engine, config)

    # 5. 创建后台任务
    engine_task = asyncio.create_task(engine.run())
    scheduler_task = asyncio.create_task(scheduler.run_scheduling_loop())

    # 6. 优雅停机处理
    loop = asyncio.get_running_loop()
    stop_event = asyncio.Event()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop_event.set)

    print("System is running. Press Ctrl+C to stop.")
    print(f"  - Workflow Engine: Active")
    print(f"  - Dynamic Scheduler: Active (strategy: {config['scheduler']['strategy']})")
    print(f"  - Metrics: http://localhost:{config['monitoring']['prometheus_port']}/metrics")
    print("\nYou can now run `python run_workload.py` to generate workload.")

    await stop_event.wait()
    print("\nShutting down...")

    # 7. 清理
    scheduler.stop()
    scheduler_task.cancel()
    engine_task.cancel()
    await asyncio.gather(scheduler_task, engine_task, return_exceptions=True)

    # 停止所有执行单元
    units = await virt_env.list_units()
    stop_tasks = [unit.stop() for unit in units.values()]
    await asyncio.gather(*stop_tasks, return_exceptions=True)

    print("Shutdown complete.")

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

"""
生成模拟工作流负载的脚本。
"""
import asyncio
import random
import sys
import uuid
from main import load_config, WorkflowEngine, DockerVirtualizationEnv

async def submit_random_workflow(engine: WorkflowEngine):
    """提交一个随机步骤的工作流"""
    steps = []
    skills = ["text_processor", "ai_inference", "data_aggregator"]
    num_steps = random.randint(2, 4)
    for i in range(num_steps):
        skill = random.choice(skills)
        input_data = f"Payload-{uuid.uuid4().hex[:6]} for step {i+1}"
        steps.append({"skill": skill, "input": input_data})
    try:
        wf_id = await engine.submit_workflow(steps)
        print(f"  Submitted workflow {wf_id} with {num_steps} steps.")
        return True
    except Exception as e:
        print(f"  Failed to submit workflow: {e}")
        return False

async def continuous_workload(duration_seconds: int = 60, request_per_second: float = 2.0):
    """持续生成负载"""
    config = load_config()
    virt_env = DockerVirtualizationEnv(config['virtualization'])
    engine = WorkflowEngine(virt_env, config['workflow']['max_queue_size'])

    # 启动引擎处理循环（独立任务）
    engine_task = asyncio.create_task(engine.run())

    interval = 1.0 / request_per_second
    end_time = asyncio.get_event_loop().time() + duration_seconds

    print(f"Starting workload for {duration_seconds} seconds at {request_per_second} req/s...")
    count = 0
    while asyncio.get_event_loop().time() < end_time:
        start = asyncio.get_event_loop().time()
        success = await submit_random_workflow(engine)
        if success:
            count += 1
        elapsed = asyncio.get_event_loop().time() - start
        await asyncio.sleep(max(0, interval - elapsed))

    print(f"Workload finished. Submitted {count} workflows.")
    engine_task.cancel()
    try:
        await engine_task
    except asyncio.CancelledError:
        pass

    # 简单清理
    units = await virt_env.list_units()
    for unit in units.values():
        await unit.stop()

if __name__ == "__main__":
    if len(sys.argv) > 1:
        duration = int(sys.argv[1])
    else:
        duration = 60
    if len(sys.argv) > 2:
        rps = float(sys.argv[2])
    else:
        rps = 2.0
    asyncio.run(continuous_workload(duration, rps))

"""
工作流引擎单元测试。
"""
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock
from workflow.engine import WorkflowEngine, WorkflowStep
from virtualization.base import VirtualizationEnv, ExecutionUnit, ResourceSpec

@pytest.mark.asyncio
async def test_workflow_submission():
    """测试工作流提交"""
    mock_env = MagicMock(spec=VirtualizationEnv)
    mock_env.create_unit = AsyncMock()
    mock_unit = MagicMock(spec=ExecutionUnit)
    mock_unit.unit_id = "test-unit"
    mock_unit.resource_spec = ResourceSpec(1.0, 512)
    mock_unit.status = "running"
    mock_unit.start = AsyncMock()
    mock_unit.execute = AsyncMock(return_value={"success": True, "output": "test output", "execution_time_ms": 50})
    mock_env.create_unit.return_value = mock_unit

    engine = WorkflowEngine(mock_env, max_queue_size=10)

    steps_def = [{"skill": "text_processor", "input": "Hello"}]
    wf_id = await engine.submit_workflow(steps_def)

    assert wf_id.startswith("wf-")
    assert engine.pending_queue.qsize() == 1
    # 注意：由于引擎处理是异步的，这里可能需要等待或直接测试提交逻辑

@pytest.mark.asyncio
async def test_skill_execution():
    """测试技能执行"""
    from workflow.skills import SKILL_REGISTRY
    # 确保技能存在
    assert "text_processor" in SKILL_REGISTRY
    skill = SKILL_REGISTRY["text_processor"]
    assert skill.name == "text_processor"
    assert "python -c" in skill.command

mkdir agent-virt-tuning && cd agent-virt-tuning

pip install -r requirements.txt

python main.py

# 默认运行60秒，每秒2个请求
    python run_workload.py
    # 自定义运行120秒，每秒5个请求
    # python run_workload.py 120 5

pytest tests/ -v

# quick_test.py
import asyncio
import sys
sys.path.insert(0, '.')
from main import load_config, WorkflowEngine, DockerVirtualizationEnv

async def test():
    config = load_config()
    virt_env = DockerVirtualizationEnv(config['virtualization'])
    engine = WorkflowEngine(virt_env, 10)
    # 启动引擎
    engine_task = asyncio.create_task(engine.run())
    # 提交一个简单工作流
    steps = [{"skill": "text_processor", "input": "Quick Test Input"}]
    wf_id = await engine.submit_workflow(steps)
    print(f"Submitted: {wf_id}")
    # 等待几秒让任务处理
    await asyncio.sleep(3)
    # 检查实例状态（简化，实际应从引擎获取）
    print("Test completed. Check main console for execution logs.")
    engine_task.cancel()
    await engine_task

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