一名工程师在凌晨2点部署了自动化报告生成系统，却在Slack上收到警报：「ConnectionError: timeout après 30 secondes — Agent validateur bloqué」。检查后发现：三个AI Agent在等待第四个Agent的状态更新，但由于网络超时，状态机陷入了死锁。这个场景揭示了多Agent系统中最常见的三个陷阱：任务分配混乱、状态共享失败、错误恢复缺失。今天，我们从头构建一个健壮的multi-agent协作系统。
为什么需要Multi-Agent架构？
当我们构建复杂的AI应用时，单一Agent往往难以同时处理数据提取、内容生成、质量验证和格式输出等异构任务。Multi-Agent架构允许每个Agent专注于特定领域，通过协作完成端到端流程。以HolySheep AI为例，其<50ms的延迟和稳定的API连接使其成为多Agent协作的理想底座。
系统架构概览
我们的系统包含四个核心组件：
- Orchestrator Agent — 负责任务分解和分配
- Researcher Agent — 负责数据收集和验证
- Writer Agent — 负责内容生成
- Validator Agent — 负责质量检查和输出
1. 核心状态共享机制
状态共享是多Agent协作的命脉。我们使用Redis作为共享状态存储，配合事件驱动架构确保各Agent实时感知系统状态。
import redis
import json
import time
from typing import Any, Dict, Optional
from dataclasses import dataclass, asdict
from enum import Enum
class AgentStatus(Enum):
IDLE = "idle"
WORKING = "working"
WAITING = "waiting"
COMPLETED = "completed"
FAILED = "failed"
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
BLOCKED = "blocked"
DONE = "done"
FAILED = "failed"
@dataclass
class SharedState:
"""共享状态结构"""
agent_id: str
status: AgentStatus
current_task: Optional[str]
progress: float # 0.0 - 1.0
result: Optional[Dict[str, Any]]
error: Optional[str]
updated_at: float
def to_json(self) -> str:
return json.dumps(asdict(self))
@classmethod
def from_json(cls, data: str) -> 'SharedState':
return cls(**json.loads(data))
class StateManager:
"""状态管理器 - 线程安全的状态共享"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.channel = "agent_state_updates"
self.pubsub = self.redis.pubsub()
self.pubsub.subscribe(self.channel)
def update_state(self, agent_id: str, state: SharedState) -> None:
"""更新Agent状态并广播"""
state.updated_at = time.time()
key = f"agent:state:{agent_id}"
# 使用事务确保原子性
pipe = self.redis.pipeline()
pipe.set(key, state.to_json())
pipe.expire(key, 3600) # 1小时过期
pipe.publish(self.channel, json.dumps({
"agent_id": agent_id,
"status": state.status.value
}))
pipe.execute()
def get_state(self, agent_id: str) -> Optional[SharedState]:
"""获取指定Agent状态"""
key = f"agent:state:{agent_id}"
data = self.redis.get(key)
return SharedState.from_json(data) if data else None
def get_all_states(self) -> Dict[str, SharedState]:
"""获取所有Agent状态"""
keys = self.redis.keys("agent:state:*")
states = {}
for key in keys:
agent_id = key.decode().split(":")[-1]
states[agent_id] = self.get_state(agent_id)
return states
def wait_for_condition(
self,
agent_id: str,
condition: callable,
timeout: float = 30.0
) -> bool:
"""等待条件满足"""
start = time.time()
while time.time() - start < timeout:
state = self.get_state(agent_id)
if state and condition(state):
return True
time.sleep(0.5)
return False2. 智能任务分配器
任务分配器需要考虑负载均衡、技能匹配和依赖关系。我们实现了一个基于优先级的分配算法。
from typing import List, Callable
from dataclasses import dataclass
import asyncio
from openai import OpenAI
from HolySheep_API import HolySheepClient # 假设的SDK
初始化HolySheep客户端
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
base_url = "https://api.holysheep.ai/v1"
@dataclass
class Agent:
id: str
name: str
capabilities: List[str]
max_concurrent: int = 2
current_load: int = 0
@dataclass
class Task:
id: str
type: str
required_capabilities: List[str]
priority: int # 1-10, 10最高
dependencies: List[str] = None
payload: Dict[str, Any] = None
def __lt__(self, other):
return self.priority < other.priority
class TaskAllocator:
"""智能任务分配器"""
def __init__(self, state_manager: StateManager):
self.state_manager = state_manager
self.agents: Dict[str, Agent] = {}
self.task_queue: asyncio.PriorityQueue = None
self._running = False
def register_agent(self, agent: Agent) -> None:
"""注册Agent"""
self.agents[agent.id] = agent
initial_state = SharedState(
agent_id=agent.id,
status=AgentStatus.IDLE,
current_task=None,
progress=0.0,
result=None,
error=None,
updated_at=time.time()
)
self.state_manager.update_state(agent.id, initial_state)
async def allocate_task(self, task: Task) -> Optional[str]:
"""分配任务到最合适的Agent"""
# 检查依赖是否满足
if task.dependencies:
deps_met = await self._check_dependencies(task.dependencies)
if not deps_met:
return None
# 筛选有能力且有空闲的Agent
candidates = [
agent for agent in self.agents.values()
if self._has_capability(agent, task.required_capabilities)
and agent.current_load < agent.max_concurrent
]
if not candidates:
return None
# 选择负载最低的Agent
best_agent = min(candidates, key=lambda a: a.current_load)
best_agent.current_load += 1
# 更新状态
state = self.state_manager.get_state(best_agent.id)
state.status = AgentStatus.WORKING
state.current_task = task.id
state.progress = 0.0
self.state_manager.update_state(best_agent.id, state)
return best_agent.id
def _has_capability(self, agent: Agent, required: List[str]) -> bool:
return all(cap in agent.capabilities for cap in required)
async def _check_dependencies(self, deps: List[str]) -> bool:
for dep_id in deps:
state = self.state_manager.get_state(dep_id)
if not state or state.status != AgentStatus.COMPLETED:
return False
return True
async def run_allocation_loop(self):
"""主分配循环"""
self._running = True
while self._running:
# 从消息队列获取任务（简化实现）
# 实际应用中应接入RabbitMQ/Kafka
task = await self._get_next_task()
if task:
agent_id = await self.allocate_task(task)
if agent_id:
await self._dispatch_task(agent_id, task)
else:
# 重新入队，稍后重试
await self._requeue_task(task)
await asyncio.sleep(0.1)
async def _get_next_task(self) -> Optional[Task]:
# 实现任务获取逻辑
pass
async def _dispatch_task(self, agent_id: str, task: Task):
# 实现任务分发
pass
async def _requeue_task(self, task: Task):
# 实现任务重入队
pass3. Agent执行引擎与HolySheep集成
每个Agent通过统一的执行引擎与HolySheep API交互。我们实现了重试机制和超时控制。
import httpx
from typing import Optional, Dict, Any
import asyncio
class AgentExecutor:
"""Agent执行引擎 - 集成HolySheep API"""
def __init__(self, state_manager: StateManager):
self.state_manager = state_manager
self.client = httpx.AsyncClient(
base_url="https://api.holysheep.ai/v1",
timeout=httpx.Timeout(60.0, connect=10.0),
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
)
self.max_retries = 3
self.retry_delay = 2.0
async def execute_task(
self,
agent_id: str,
task: Task,
system_prompt: str,
user_message: str
) -> Dict[str, Any]:
"""执行任务并处理错误恢复"""
state = self.state_manager.get_state(agent_id)
for attempt in range(self.max_retries):
try:
# 调用HolySheep API
response = await self._call_holysheep(
system_prompt=system_prompt,
user_message=user_message
)
# 更新成功状态
state.status = AgentStatus.COMPLETED
state.progress = 1.0
state.result = response
self.state_manager.update_state(agent_id, state)
return response
except httpx.TimeoutException as e:
# 超时处理
state.error = f"TimeoutError: {str(e)}"
state.status = AgentStatus.WAITING
self.state_manager.update_state(agent_id, state)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delay * (attempt + 1))
continue
else:
state.status = AgentStatus.FAILED
self.state_manager.update_state(agent_id, state)
raise
except httpx.HTTPStatusError as e:
# HTTP错误处理
if e.response.status_code == 401:
raise Exception("API密钥无效或已过期") from e
elif e.response.status_code == 429:
# 速率限制 - 指数退避
await asyncio.sleep(2 ** attempt)
continue
else:
state.error = f"HTTP {e.response.status_code}: {str(e)}"
state.status = AgentStatus.FAILED
self.state_manager.update_state(agent_id, state)
raise
except httpx.ConnectError as e:
# 连接错误 - 可能是网络问题或API不可用
state.error = f"ConnectionError: {str(e)}"
self.state_manager.update_state(agent_id, state)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delay)
continue
else:
state.status = AgentStatus.FAILED
self.state_manager.update_state(agent_id, state)
raise
async def _call_holysheep(
self,
system_prompt: str,
user_message: str,
model: str = "gpt-4o" # 或 "claude-3-5-sonnet", "deepseek-v3"
) -> Dict[str, Any]:
"""调用HolySheep API"""
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
"temperature": 0.7,
"max_tokens": 4096
}
response = await self.client.post("/chat/completions", json=payload)
response.raise_for_status()
data = response.json()
return {
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {}),
"model": data.get("model")
}
async def execute_with_fallback(
self,
agent_id: str,
task: Task,
primary_model: str,
fallback_model: str
) -> Dict[str, Any]:
"""使用降级模型的执行"""
try:
return await self.execute_task(
agent_id, task,
self._get_system_prompt(task),
self._get_user_message(task),
model=primary_model
)
except Exception as e:
print(f"主模型 {primary_model} 失败，切换到 {fallback_model}")
return await self.execute_task(
agent_id, task,
self._get_system_prompt(task),
self._get_user_message(task),
model=fallback_model
)4. 错误恢复与状态回滚
健壮的错误恢复机制是多Agent系统稳定运行的关键。我们实现了基于检查点的恢复和事务性回滚。
from typing import Optional, Callable, Any
from enum import Enum
import traceback
from datetime import datetime
class RecoveryStrategy(Enum):
RETRY = "retry"
FALLBACK = "fallback"
ROLLBACK = "rollback"
SKIP = "skip"
ESCALATE = "escalate"
class CheckpointManager:
"""检查点管理器 - 支持任务恢复"""
def __init__(self, state_manager: StateManager):
self.state_manager = state_manager
self.checkpoints: Dict[str, List[Dict]] = {}
def save_checkpoint(
self,
task_id: str,
agent_id: str,
step: int,
data: Dict[str, Any]
) -> None:
"""保存检查点"""
key = f"checkpoint:{task_id}:{agent_id}"
checkpoint = {
"step": step,
"data": data,
"timestamp": datetime.utcnow().isoformat()
}
# 追加到列表
self.state_manager.redis.rpush(key, json.dumps(checkpoint))
self.state_manager.redis.expire(key, 86400) # 24小时
def get_latest_checkpoint(
self,
task_id: str,
agent_id: str
) -> Optional[Dict]:
"""获取最新检查点"""
key = f"checkpoint:{task_id}:{agent_id}"
checkpoints = self.state_manager.redis.lrange(key, -1, -1)
if checkpoints:
return json.loads(checkpoints[0])
return None
def clear_checkpoints(self, task_id: str, agent_id: str) -> None:
"""清除检查点"""
key = f"checkpoint:{task_id}:{agent_id}"
self.state_manager.redis.delete(key)
class ErrorRecoveryManager:
"""错误恢复管理器"""
def __init__(
self,
state_manager: StateManager,
checkpoint_manager: CheckpointManager
):
self.state_manager = state_manager
self.checkpoint_manager = checkpoint_manager
self.error_handlers: Dict[str, Callable] = {}
def register_handler(
self,
error_type: str,
strategy: RecoveryStrategy,
handler: Callable
) -> None:
"""注册错误处理器"""
self.error_handlers[error_type] = {
"strategy": strategy,
"handler": handler
}
async def handle_error(
self,
agent_id: str,
task_id: str,
error: Exception
) -> bool:
"""处理错误并执行恢复"""
error_type = type(error).__name__
state = self.state_manager.get_state(agent_id)
print(f"[{agent_id}] 捕获错误: {error_type} - {str(error)}")
print(f"堆栈跟踪:\n{traceback.format_exc()}")
# 检查是否有注册的处理程序
handler_info = self.error_handlers.get(error_type)
if handler_info:
strategy = handler_info["strategy"]
handler = handler_info["handler"]
if strategy == RecoveryStrategy.ROLLBACK:
return await self._rollback_to_checkpoint(
agent_id, task_id, state
)
elif strategy == RecoveryStrategy.FALLBACK:
return await self._execute_fallback(handler, agent_id, task_id)
elif strategy == RecoveryStrategy.SKIP:
return await self._skip_task(agent_id, task_id)
elif strategy == RecoveryStrategy.ESCALATE:
return await self._escalate_error(agent_id, task_id, error)
# 默认策略：标记失败并记录
state.status = AgentStatus.FAILED
state.error = f"{error_type}: {str(error)}"
self.state_manager.update_state(agent_id
Ressources connexes
Articles connexes