多智能体协作系统：2026年企业级AI应用的核心架构范式

# Python示例：使用FastAPI构建多Agent系统的API网关
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional, List, Dict, Any
import asyncio
from datetime import datetime
import hashlib

app = FastAPI(title="Multi-Agent Collaboration System API")
security = HTTPBearer()

# 请求模型
class AgentRequest(BaseModel):
    session_id: str
    message: str
    context: Optional[Dict[str, Any]] = None
    agent_types: Optional[List[str]] = None  # 指定使用的Agent类型

class AgentResponse(BaseModel):
    session_id: str
    message: str
    agent_id: str
    timestamp: str
    metadata: Optional[Dict[str, Any]] = None

# 简单的内存会话存储（生产环境应使用Redis）
sessions: Dict[str, Dict[str, Any]] = {}

# 认证依赖
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    # 在实际应用中，这里应该验证JWT或OAuth token
    if not token.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid token format")
    return {"user_id": hashlib.md5(token.encode()).hexdigest()[:8]}

@app.post("/api/v1/agent/invoke", response_model=AgentResponse)
async def invoke_agent(
    request: AgentRequest,
    auth: dict = Depends(verify_token)
):
    """
    调用多Agent系统处理用户请求
    """
    # 检查会话状态
    session = sessions.get(request.session_id)
    if session is None:
        session = {
            "created_at": datetime.now().isoformat(),
            "history": [],
            "agent_states": {}
        }
        sessions[request.session_id] = session
    
    # 将用户消息添加到历史
    session["history"].append({
        "role": "user",
        "content": request.message,
        "timestamp": datetime.now().isoformat()
    })
    
    # 这里应该调用Orchestrator进行任务规划和Agent调度
    # 为了示例简化，直接返回响应
    response = AgentResponse(
        session_id=request.session_id,
        message=f"Processed by Multi-Agent System: {request.message}",
        agent_id="orchestrator",
        timestamp=datetime.now().isoformat(),
        metadata={
            "agent_count": 3,
            "processing_time_ms": 150
        }
    )
    
    session["history"].append({
        "role": "assistant",
        "content": response.message,
        "timestamp": response.timestamp
    })
    
    return response

@app.get("/api/v1/session/{session_id}/history")
async def get_session_history(
    session_id: str,
    auth: dict = Depends(verify_token)
):
    """获取会话历史"""
    session = sessions.get(session_id)
    if session is None:
        raise HTTPException(status_code=404, detail="Session not found")
    return session["history"]

@app.delete("/api/v1/session/{session_id}")
async def delete_session(
    session_id: str,
    auth: dict = Depends(verify_token)
):
    """删除会话"""
    if session_id in sessions:
        del sessions[session_id]
    return {"status": "deleted", "session_id": session_id}

# Python示例：任务规划器实现
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Set
from enum import Enum
import json

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

@dataclass
class Task:
    id: str
    name: str
    description: str
    required_skills: List[str]
    estimated_duration: int  # 秒
    dependencies: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    assigned_agent: Optional[str] = None
    result: Optional[Any] = None
    error: Optional[str] = None

@dataclass
class ExecutionPlan:
    tasks: List[Task]
    parallel_groups: List[List[str]]  # 可并行执行的任务组
    total_duration: int
    critical_path: List[str]  # 关键路径

class TaskPlanner:
    """任务规划器：负责将复杂任务分解为可执行的子任务"""
    
    def __init__(self):
        # 内置的任务模板
        self.task_templates = {
            "code_development": [
                Task(id="req_analysis", name="需求分析", 
                     description="分析用户需求，输出需求规格说明书",
                     required_skills=["分析", "文档"], estimated_duration=300),
                Task(id="code_design", name="架构设计",
                     description="设计系统架构和数据模型",
                     required_skills=["架构设计", "UML"], estimated_duration=600,
                     dependencies=["req_analysis"]),
                Task(id="code_write", name="代码编写",
                     description="按照设计文档编写代码",
                     required_skills=["编程"], estimated_duration=1800,
                     dependencies=["code_design"]),
                Task(id="code_review", name="代码审查",
                     description="审查代码质量和安全性",
                     required_skills=["代码审查"], estimated_duration=600,
                     dependencies=["code_write"]),
                Task(id="unit_test", name="单元测试",
                     description="编写和执行单元测试",
                     required_skills=["测试"], estimated_duration=900,
                     dependencies=["code_write"]),
                Task(id="integration_test", name="集成测试",
                     description="执行集成测试",
                     required_skills=["测试"], estimated_duration=600,
                     dependencies=["code_review", "unit_test"]),
            ],
            "data_analysis": [
                Task(id="data_collect", name="数据采集",
                     description="从多个数据源采集数据",
                     required_skills=["数据采集"], estimated_duration=300),
                Task(id="data_clean", name="数据清洗",
                     description="清洗和预处理数据",
                     required_skills=["数据处理"], estimated_duration=600,
                     dependencies=["data_collect"]),
                Task(id="data_explore", name="探索性分析",
                     description="进行探索性数据分析",
                     required_skills=["分析"], estimated_duration=600,
                     dependencies=["data_clean"]),
                Task(id="model_build", name="建模分析",
                     description="构建分析模型",
                     required_skills=["建模"], estimated_duration=1200,
                     dependencies=["data_explore"]),
                Task(id="report_gen", name="报告生成",
                     description="生成分析报告和可视化",
                     required_skills=["可视化", "文档"], estimated_duration=600,
                     dependencies=["model_build"]),
            ]
        }
    
    def parse_user_request(self, message: str) -> str:
        """根据用户消息推断任务类型"""
        message_lower = message.lower()
        
        if any(keyword in message_lower for keyword in ["开发", "编写", "创建", "实现", "代码"]):
            return "code_development"
        elif any(keyword in message_lower for keyword in ["分析", "数据", "报表", "统计", "挖掘"]):
            return "data_analysis"
        elif any(keyword in message_lower for keyword in ["搜索", "查找", "查询", "研究"]):
            return "research"
        else:
            return "general"
    
    def create_execution_plan(self, task_type: str, context: Optional[Dict] = None) -> ExecutionPlan:
        """根据任务类型创建执行计划"""
        if task_type not in self.task_templates:
            # 通用任务：创建一个默认任务
            tasks = [Task(
                id="general_task",
                name="通用任务",
                description="处理用户请求",
                required_skills=["通用"],
                estimated_duration=300
            )]
            return ExecutionPlan(
                tasks=tasks,
                parallel_groups=[["general_task"]],
                total_duration=300,
                critical_path=["general_task"]
            )
        
        tasks = self.task_templates[task_type]
        
        # 根据上下文调整任务参数
        if context:
            for task in tasks:
                # 可以根据上下文添加额外的任务或修改参数
                pass
        
        # 计算并行组（根据依赖关系）
        parallel_groups = self._compute_parallel_groups(tasks)
        
        # 计算关键路径
        critical_path = self._compute_critical_path(tasks)
        
        # 计算总工期
        total_duration = sum(t.estimated_duration for t in critical_path)
        
        return ExecutionPlan(
            tasks=tasks,
            parallel_groups=parallel_groups,
            total_duration=total_duration,
            critical_path=critical_path
        )
    
    def _compute_parallel_groups(self, tasks: List[Task]) -> List[List[str]]:
        """计算可并行执行的任务组"""
        # 简化实现：按依赖层次分组
        groups = []
        remaining = set(t.id for t in tasks)
        completed = set()
        
        while remaining:
            # 找出所有依赖都已完成的任务
            current_group = []
            for task_id in list(remaining):
                task = next(t for t in tasks if t.id == task_id)
                if all(dep in completed for dep in task.dependencies):
                    current_group.append(task_id)
            
            if not current_group:
                # 防止死循环
                current_group = [min(remaining, key=lambda x: len(
                    next(t for t in tasks if t.id == x).dependencies
                ))]
            
            groups.append(current_group)
            completed.update(current_group)
            remaining -= set(current_group)
        
        return groups
    
    def _compute_critical_path(self, tasks: List[Task]) -> List[str]:
        """计算关键路径（最长依赖链）"""
        # 简化实现：返回按依赖顺序排列的任务链
        critical = []
        remaining = set(t.id for t in tasks)
        completed = {}
        
        while remaining:
            for task_id in list(remaining):
                task = next(t for t in tasks if t.id == task_id)
                if all(dep in completed for dep in task.dependencies):
                    # 这个任务可以被执行，加入关键路径
                    if not critical or task.dependencies:
                        critical.append(task_id)
                    completed[task_id] = task
                    remaining.remove(task_id)
                    break
        
        return critical

# 使用示例
planner = TaskPlanner()
plan = planner.create_execution_plan("code_development")
print(f"执行计划：")
print(f"  总工期：{plan.total_duration}秒")
print(f"  任务数量：{len(plan.tasks)}")
print(f"  关键路径：{' -> '.join(plan.critical_path)}")
print(f"  并行组：{plan.parallel_groups}")

// Go示例：多Agent系统的Agent定义和调度器
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"
)

// AgentType Agent类型
type AgentType string

const (
	AgentTypePlanner  AgentType = "planner"
	AgentTypeCoder    AgentType = "coder"
	AgentTypeSearch   AgentType = "search"
	AgentTypeData     AgentType = "data"
	AgentTypeWriter   AgentType = "writer"
	AgentTypeReviewer AgentType = "reviewer"
)

// Task 任务定义
type Task struct {
	ID              string                 `json:"id"`
	Name            string                 `json:"name"`
	RequiredAgent   AgentType              `json:"required_agent"`
	Input           map[string]interface{} `json:"input"`
	Dependencies    []string               `json:"dependencies,omitempty"`
	Status          string                 `json:"status"`
	Result          interface{}            `json:"result,omitempty"`
	Error           string                 `json:"error,omitempty"`
	AssignedAgentID string                 `json:"assigned_agent_id,omitempty"`
}

// AgentResult Agent执行结果
type AgentResult struct {
	TaskID    string      `json:"task_id"`
	AgentID   string      `json:"agent_id"`
	Success   bool        `json:"success"`
	Output    interface{} `json:"output,omitempty"`
	Error     string      `json:"error,omitempty"`
	Duration  time.Duration `json:"duration"`
}

// Agent Agent接口
type Agent interface {
	GetType() AgentType
	GetCapabilities() []string
	Execute(ctx context.Context, task *Task) (*AgentResult, error)
}

// BaseAgent Agent基类
type BaseAgent struct {
	ID          string
	AgentType   AgentType
	Capabilities []string
}

func (b *BaseAgent) GetType() AgentType {
	return b.AgentType
}

func (b *BaseAgent) GetCapabilities() []string {
	return b.Capabilities
}

// PlannerAgent 规划Agent
type PlannerAgent struct {
	BaseAgent
}

func NewPlannerAgent() *PlannerAgent {
	return &PlannerAgent{
		BaseAgent: BaseAgent{
			ID:        "planner-001",
			AgentType: AgentTypePlanner,
			Capabilities: []string{"task_decomposition", "dependency_analysis", "resource_planning"},
		},
	}
}

func (a *PlannerAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	// 模拟任务规划过程
	inputJSON, _ := json.Marshal(task.Input)
	fmt.Printf("[PlannerAgent] Planning task: %s, input: %s\n", task.Name, string(inputJSON))
	
	// 任务分解逻辑
	subTasks := []map[string]interface{}{
		{"id": "sub_1", "name": "数据收集", "agent": "search"},
		{"id": "sub_2", "name": "数据分析", "agent": "data", "depends_on": []string{"sub_1"}},
		{"id": "sub_3", "name": "报告撰写", "agent": "writer", "depends_on": []string{"sub_2"}},
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"sub_tasks": subTasks, "execution_order": []string{"sub_1", "sub_2", "sub_3"}},
		Duration: time.Since(start),
	}, nil
}

// CoderAgent 代码Agent
type CoderAgent struct {
	BaseAgent
	languageModels map[string]bool
}

func NewCoderAgent() *CoderAgent {
	return &CoderAgent{
		BaseAgent: BaseAgent{
			ID:          "coder-001",
			AgentType:   AgentTypeCoder,
			Capabilities: []string{"code_generation", "code_review", "refactoring"},
		},
		languageModels: map[string]bool{
			"python": true,
			"go":     true,
			"java":   true,
			"rust":   true,
		},
	}
}

func (a *CoderAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	// 检查语言支持
	language, _ := task.Input["language"].(string)
	if !a.languageModels[language] {
		return &AgentResult{
			TaskID:   task.ID,
			AgentID:  a.ID,
			Success:  false,
			Error:    fmt.Sprintf("Language %s not supported", language),
			Duration: time.Since(start),
		}, nil
	}
	
	// 模拟代码生成
	code := fmt.Sprintf("// Generated %s code for: %s\nfunc main() {\n    // TODO: implement %s\n}\n", 
		language, task.Name, task.Name)
	
	fmt.Printf("[CoderAgent] Generated %s code for task: %s\n", language, task.Name)
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"language": language, "code": code, "lines": 5},
		Duration: time.Since(start),
	}, nil
}

// SearchAgent 搜索Agent
type SearchAgent struct {
	BaseAgent
}

func NewSearchAgent() *SearchAgent {
	return &SearchAgent{
		BaseAgent: BaseAgent{
			ID:          "search-001",
			AgentType:   AgentTypeSearch,
			Capabilities: []string{"web_search", "document_search", "knowledge_retrieval"},
		},
	}
}

func (a *SearchAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	query, _ := task.Input["query"].(string)
	fmt.Printf("[SearchAgent] Searching for: %s\n", query)
	
	// 模拟搜索结果
	results := []map[string]string{
		{"title": "Result 1", "url": "https://example.com/1", "snippet": "Relevant information..."},
		{"title": "Result 2", "url": "https://example.com/2", "snippet": "More information..."},
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"query": query, "results": results, "total": len(results)},
		Duration: time.Since(start),
	}, nil
}

// DataAgent 数据分析Agent
type DataAgent struct {
	BaseAgent
}

func NewDataAgent() *DataAgent {
	return &DataAgent{
		BaseAgent: BaseAgent{
			ID:          "data-001",
			AgentType:   AgentTypeData,
			Capabilities: []string{"data_processing", "统计分析", "可视化"},
		},
	}
}

func (a *DataAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	// 模拟数据分析
	data, _ := json.Marshal(task.Input)
	fmt.Printf("[DataAgent] Processing data: %s\n", string(data))
	
	analysis := map[string]interface{}{
		"record_count":   1000,
		"summary":        "数据总体呈上升趋势",
		"anomalies":      []string{},
		"recommendations": []string{"建议优化A指标", "关注B类数据波动"},
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   analysis,
		Duration: time.Since(start),
	}, nil
}

// WriterAgent 写作Agent
type WriterAgent struct {
	BaseAgent
}

func NewWriterAgent() *WriterAgent {
	return &WriterAgent{
		BaseAgent: BaseAgent{
			ID:          "writer-001",
			AgentType:   AgentTypeWriter,
			Capabilities: []string{"document_writing", "report_generation", "content_creation"},
		},
	}
}

func (a *WriterAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	content, _ := task.Input["content"].(string)
	style, _ := task.Input["style"].(string)
	
	fmt.Printf("[WriterAgent] Writing content in %s style\n", style)
	
	document := fmt.Sprintf("# %s\n\n%s\n\n## 结论\n\n基于以上分析，生成专业报告。", 
		task.Name, content)
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"document": document, "word_count": len(document)},
		Duration: time.Since(start),
	}, nil
}

// ReviewerAgent 审核Agent
type ReviewerAgent struct {
	BaseAgent
}

func NewReviewerAgent() *ReviewerAgent {
	return &ReviewerAgent{
		BaseAgent: BaseAgent{
			ID:          "reviewer-001",
			AgentType:   AgentTypeReviewer,
			Capabilities: []string{"code_review", "quality_check", "compliance_verification"},
		},
	}
}

func (a *ReviewerAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	content, _ := json.Marshal(task.Input)
	fmt.Printf("[ReviewerAgent] Reviewing content: %s\n", string(content))
	
	review := map[string]interface{}{
		"quality_score":   85,
		"issues":          []string{},
		"suggestions":     []string{"可进一步优化性能"},
		"approved":        true,
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   review,
		Duration: time.Since(start),
	}, nil
}

// AgentRegistry Agent注册表
type AgentRegistry struct {
	mu     sync.RWMutex
	agents map[AgentType]Agent
}

func NewAgentRegistry() *AgentRegistry {
	registry := &AgentRegistry{
		agents: make(map[AgentType]Agent),
	}
	
	// 注册内置Agent
	registry.Register(NewPlannerAgent())
	registry.Register(NewCoderAgent())
	registry.Register(NewSearchAgent())
	registry.Register(NewDataAgent())
	registry.Register(NewWriterAgent())
	registry.Register(NewReviewerAgent())
	
	return registry
}

func (r *AgentRegistry) Register(agent Agent) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.agents[agent.GetType()] = agent
}

func (r *AgentRegistry) Get(agentType AgentType) (Agent, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	agent, ok := r.agents[agentType]
	return agent, ok
}

func (r *AgentRegistry) List() []Agent {
	r.mu.RLock()
	defer r.mu.RUnlock()
	
	agents := make([]Agent, 0, len(r.agents))
	for _, agent := range r.agents {
		agents = append(agents, agent)
	}
	return agents
}

// AgentScheduler Agent调度器
type AgentScheduler struct {
	registry *AgentRegistry
	mu       sync.Mutex
}

func NewAgentScheduler(registry *AgentRegistry) *AgentScheduler {
	return &AgentScheduler{
		registry: registry,
	}
}

func (s *AgentScheduler) Schedule(task *Task) (Agent, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	agent, ok := s.registry.Get(task.RequiredAgent)
	if !ok {
		return nil, fmt.Errorf("no agent available for type: %s", task.RequiredAgent)
	}
	
	return agent, nil
}

// MultiAgentOrchestrator 多Agent编排器
type MultiAgentOrchestrator struct {
	registry  *AgentRegistry
	scheduler *AgentScheduler
}

func NewMultiAgentOrchestrator() *MultiAgentOrchestrator {
	registry := NewAgentRegistry()
	return &MultiAgentOrchestrator{
		registry:  registry,
		scheduler: NewAgentScheduler(registry),
	}
}

func (o *MultiAgentOrchestrator) ExecuteTask(ctx context.Context, task *Task) (*AgentResult, error) {
	// 调度合适的Agent
	agent, err := o.scheduler.Schedule(task)
	if err != nil {
		return nil, err
	}
	
	// 执行任务
	return agent.Execute(ctx, task)
}

func (o *MultiAgentOrchestrator) ExecuteParallelTasks(ctx context.Context, tasks []*Task) ([]*AgentResult, error) {
	var wg sync.WaitGroup
	results := make([]*AgentResult, len(tasks))
	errors := make([]error, len(tasks))
	
	for i, task := range tasks {
		wg.Add(1)
		go func(idx int, t *Task) {
			defer wg.Done()
			
			result, err := o.ExecuteTask(ctx, t)
			results[idx] = result
			errors[idx] = err
		}(i, task)
	}
	
	wg.Wait()
	
	// 检查是否有错误
	for _, err := range errors {
		if err != nil {
			return results, err
		}
	}
	
	return results, nil
}

func main() {
	// 创建编排器
	orchestrator := NewMultiAgentOrchestrator()
	ctx := context.Background()
	
	// 创建任务
	tasks := []*Task{
		{
			ID:            "task-001",
			Name:          "搜索信息",
			RequiredAgent: AgentTypeSearch,
			Input:         map[string]interface{}{"query": "最新AI技术发展"},
		},
		{
			ID:            "task-002",
			Name:          "数据分析",
			RequiredAgent: AgentTypeData,
			Input:         map[string]interface{}{"data": []float64{1.1, 2.2, 3.3}},
		},
		{
			ID:            "task-003",
			Name:          "生成代码",
			RequiredAgent: AgentTypeCoder,
			Input:         map[string]interface{}{"language": "python", "task": "排序算法"},
		},
	}
	
	// 并行执行任务
	fmt.Println("开始并行执行任务...")
	results, err := orchestrator.ExecuteParallelTasks(ctx, tasks)
	if err != nil {
		fmt.Printf("执行出错: %v\n", err)
		return
	}
	
	// 输出结果
	fmt.Println("\n执行结果:")
	for _, result := range results {
		if result.Success {
			outputJSON, _ := json.MarshalIndent(result.Output, "", "  ")
			fmt.Printf("  Task %s: 成功 (耗时: %v)\n  输出: %s\n", 
				result.TaskID, result.Duration, string(outputJSON))
		} else {
			fmt.Printf("  Task %s: 失败 - %s\n", result.TaskID, result.Error)
		}
	}
}

# Python示例：MCP客户端实现
import json
import asyncio
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum

class MCPErrorCode(Enum):
    INVALID_REQUEST = -32600
    METHOD_NOT_FOUND = -32601
    INVALID_PARAMS = -32602
    INTERNAL_ERROR = -32603

@dataclass
class MCPRequest:
    jsonrpc: str = "2.0"
    id: Optional[int] = None
    method: str = ""
    params: Dict[str, Any] = field(default_factory=dict)

@dataclass
class MCPResponse:
    jsonrpc: str = "2.0"
    id: Optional[int] = None
    result: Optional[Any] = None
    error: Optional[Dict[str, Any]] = None

@dataclass
class MCPTool:
    name: str
    description: str
    inputSchema: Dict[str, Any]

@dataclass
class MCPToolResult:
    content: List[Dict[str, Any]]
    isError: bool = False

class MCPClient:
    """MCP客户端实现"""
    
    def __init__(self, server_url: str, api_key: Optional[str] = None):
        self.server_url = server_url
        self.api_key = api_key
        self.tools: Dict[str, MCPTool] = {}
        self._request_id = 0
    
    async def initialize(self) -> Dict[str, Any]:
        """初始化连接，获取服务器能力"""
        # 发送initialize请求
        request = MCPRequest(
            id=self._next_id(),
            method="initialize",
            params={
                "protocolVersion": "2024-11-05",
                "capabilities": {
                    "roots": {"listChanged": True},
                    "sampling": {}
                },
                "clientInfo": {
                    "name": "multi-agent-client",
                    "version": "1.0.0"
                }
            }
        )
        
        # 在实际实现中，这里会发送HTTP请求到MCP服务器
        # response = await self._send_request(request)
        
        # 模拟响应
        response = {
            "protocolVersion": "2024-11-05",
            "serverInfo": {
                "name": "example-mcp-server",
                "version": "1.0.0"
            },
            "capabilities": {
                "tools": {"listChanged": True}
            },
            "instructions": "Example MCP Server for demonstration"
        }
        
        # 获取可用工具列表
        await self.list_tools()
        
        return response
    
    async def list_tools(self) -> List[MCPTool]:
        """列出所有可用工具"""
        request = MCPRequest(
            id=self._next_id(),
            method="tools/list"
        )
        
        # 模拟响应
        tool_list = [
            MCPTool(
                name="search_web",
                description="Search the web for information",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Search query"},
                        "limit": {"type": "integer", "description": "Max results"}
                    },
                    "required": ["query"]
                }
            ),
            MCPTool(
                name="read_file",
                description="Read contents of a file",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "File path"},
                        "encoding": {"type": "string", "default": "utf-8"}
                    },
                    "required": ["path"]
                }
            ),
            MCPTool(
                name="execute_code",
                description="Execute code in a sandboxed environment",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "language": {"type": "string", "enum": ["python", "javascript", "bash"]},
                        "code": {"type": "string"}
                    },
                    "required": ["language", "code"]
                }
            ),
        ]
        
        self.tools = {tool.name: tool for tool in tool_list}
        return tool_list
    
    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> MCPToolResult:
        """调用指定工具"""
        if tool_name not in self.tools:
            return MCPToolResult(
                content=[{"type": "text", "text": f"Tool '{tool_name}' not found"}],
                isError=True
            )
        
        request = MCPRequest(
            id=self._next_id(),
            method="tools/call",
            params={
                "name": tool_name,
                "arguments": arguments
            }
        )
        
        # 模拟工具执行
        if tool_name == "search_web":
            results = [
                {"type": "text", "text": f"Search results for '{arguments.get('query', '')}':"},
                {"type": "text", "text": "1. Result A - relevant information about the topic"},
                {"type": "text", "text": "2. Result B - another relevant source"},
            ]
            return MCPToolResult(content=results)
        
        elif tool_name == "read_file":
            return MCPToolResult(
                content=[{"type": "text", "text": f"File content from {arguments.get('path', '')}:\n[simulated content]"}]
            )
        
        elif tool_name == "execute_code":
            return MCPToolResult(
                content=[{"type": "text", "text": f"Executed {arguments.get('language')} code successfully"}]
            )
        
        return MCPToolResult(content=[{"type": "text", "text": "Unknown tool"}], isError=True)
    
    def _next_id(self) -> int:
        self._request_id += 1
        return self._request_id
    
    async def close(self):
        """关闭连接"""
        # 清理资源
        self.tools.clear()


# 使用示例
async def main():
    # 创建MCP客户端
    client = MCPClient("https://mcp-server.example.com", api_key="demo-key")
    
    # 初始化
    info = await client.initialize()
    print(f"Connected to {info['serverInfo']['name']} v{info['serverInfo']['version']}")
    
    # 列出可用工具
    tools = await client.list_tools()
    print(f"\nAvailable tools ({len(tools)}):")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")
    
    # 调用工具
    result = await client.call_tool("search_web", {"query": "multi-agent AI systems"})
    print(f"\nTool result: {result.content}")
    
    # 关闭连接
    await client.close()

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

# Python示例：A2A协议实现
import asyncio
import json
import uuid
from typing import Dict, List, Callable, Any, Optional
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class MessageType(Enum):
    TASK_REQUEST = "task_request"
    TASK_RESPONSE = "task_response"
    TASK_UPDATE = "task_update"
    TASK_COMPLETE = "task_complete"
    TASK_FAILED = "task_failed"
    HEARTBEAT = "heartbeat"
    AGENT_REGISTER = "agent_register"
    AGENT_DISCOVER = "agent_discover"


@dataclass
class A2AMessage:
    id: str
    type: MessageType
    sender_id: str
    receiver_id: Optional[str]  # None表示广播
    task_id: Optional[str]
    payload: Dict[str, Any]
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    correlation_id: Optional[str] = None


@dataclass
class AgentCapability:
    name: str
    description: str
    input_schema: Dict[str, Any]
    output_schema: Dict[str, Any]


@dataclass
class AgentInfo:
    id: str
    name: str
    capabilities: List[AgentCapability]
    status: str = "online"
    endpoint: str = ""


class A2AMessageBus:
    """A2A消息总线 - Agent间通信的核心基础设施"""
    
    def __init__(self):
        self.agents: Dict[str, AgentInfo] = {}
        self.subscribers: Dict[MessageType, List[Callable]] = {}
        self.message_queue: asyncio.Queue = asyncio.Queue()
        self.running = False
        self._message_history: List[A2AMessage] = []
    
    def register_agent(self, agent: AgentInfo):
        """注册Agent到消息总线"""
        self.agents[agent.id] = agent
        logger.info(f"Agent registered: {agent.id} ({agent.name})")
        
        # 广播新Agent注册消息
        self._broadcast(A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.AGENT_REGISTER,
            sender_id="message_bus",
            receiver_id=None,
            task_id=None,
            payload={"agent": asdict(agent)}
        ))
    
    def discover_agents(self, capability: Optional[str] = None) -> List[AgentInfo]:
        """发现具有特定能力的Agent"""
        if capability is None:
            return list(self.agents.values())
        
        return [
            agent for agent in self.agents.values()
            if any(cap.name == capability for cap in agent.capabilities)
        ]
    
    def subscribe(self, message_type: MessageType, callback: Callable):
        """订阅特定类型的消息"""
        if message_type not in self.subscribers:
            self.subscribers[message_type] = []
        self.subscribers[message_type].append(callback)
    
    def unsubscribe(self, message_type: MessageType, callback: Callable):
        """取消订阅"""
        if message_type in self.subscribers:
            self.subscribers[message_type].remove(callback)
    
    async def send_message(self, message: A2AMessage) -> bool:
        """发送消息"""
        self._message_history.append(message)
        
        # 如果有指定的接收者，直接投递给接收者
        if message.receiver_id and message.receiver_id in self.agents:
            await self._deliver_to_agent(message.receiver_id, message)
            return True
        
        # 否则广播
        self._broadcast(message)
        return True
    
    def _broadcast(self, message: A2AMessage):
        """广播消息给所有订阅者"""
        if message.type in self.subscribers:
            for callback in self.subscribers[message.type]:
                try:
                    callback(message)
                except Exception as e:
                    logger.error(f"Error in subscriber callback: {e}")
    
    async def _deliver_to_agent(self, agent_id: str, message: A2AMessage):
        """将消息投递给特定Agent"""
        # 在实际实现中，这里会将消息放入Agent的收件箱
        logger.info(f"Delivering message {message.id} to agent {agent_id}")
    
    async def get_task_status(self, task_id: str) -> List[A2AMessage]:
        """获取特定任务的所有消息"""
        return [msg for msg in self._message_history if msg.task_id == task_id]
    
    async def start(self):
        """启动消息总线"""
        self.running = True
        logger.info("A2A Message Bus started")
        
        # 启动消息处理循环
        while self.running:
            try:
                # 处理消息队列
                message = await self.message_queue.get()
                await self._process_message(message)
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error(f"Error processing message: {e}")
    
    async def stop(self):
        """停止消息总线"""
        self.running = False
        logger.info("A2A Message Bus stopped")
    
    async def _process_message(self, message: A2AMessage):
        """处理消息"""
        self._broadcast(message)


class BaseA2AAgent:
    """支持A2A协议的Agent基类"""
    
    def __init__(self, agent_id: str, name: str, capabilities: List[AgentCapability], message_bus: A2AMessageBus):
        self.agent_id = agent_id
        self.name = name
        self.capabilities = capabilities
        self.message_bus = message_bus
        self.inbox: asyncio.Queue = asyncio.Queue()
        self.task_results: Dict[str, Any] = {}
        self._running = False
    
    @property
    def agent_info(self) -> AgentInfo:
        return AgentInfo(
            id=self.agent_id,
            name=self.name,
            capabilities=self.capabilities,
            status="online",
            endpoint=f"agent://{self.agent_id}"
        )
    
    async def register(self):
        """注册到消息总线"""
        self.message_bus.register_agent(self.agent_info)
    
    async def send_task_request(self, receiver_id: str, task_id: str, task_data: Dict[str, Any]) -> str:
        """发送任务请求"""
        message = A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.TASK_REQUEST,
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            task_id=task_id,
            payload=task_data,
            correlation_id=task_id
        )
        await self.message_bus.send_message(message)
        return message.id
    
    async def send_task_response(self, receiver_id: str, task_id: str, result: Any) -> str:
        """发送任务响应"""
        message = A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.TASK_RESPONSE,
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            task_id=task_id,
            payload={"result": result}
        )
        await self.message_bus.send_message(message)
        return message.id
    
    async def send_task_update(self, receiver_id: str, task_id: str, progress: int, status: str) -> str:
        """发送任务更新"""
        message = A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.TASK_UPDATE,
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            task_id=task_id,
            payload={"progress": progress, "status": status}
        )
        await self.message_bus.send_message(message)
        return message.id
    
    async def handle_message(self, message: A2AMessage):
        """处理接收到的消息"""
        if message.type == MessageType.TASK_REQUEST:
            result = await self.execute_task(message)
            await self.send_task_response(message.sender_id, message.task_id, result)
        
        elif message.type == MessageType.TASK_COMPLETE:
            self.task_results[message.task_id] = message.payload.get("result")
        
        elif message.type == MessageType.AGENT_DISCOVER:
            # 响应Agent发现请求
            response = A2AMessage(
                id=str(uuid.uuid4()),
                type=MessageType.AGENT_REGISTER,
                sender_id=self.agent_id,
                receiver_id=message.sender_id,
                task_id=None,
                payload={"agent": asdict(self.agent_info)}
            )
            await self.message_bus.send_message(response)
    
    async def execute_task(self, message: A2AMessage) -> Any:
        """执行任务 - 子类需要实现"""
        raise NotImplementedError("Subclasses must implement execute_task")
    
    async def start(self):
        """启动Agent"""
        self._running = True
        await self.register()
        logger.info(f"Agent {self.agent_id} started")
    
    async def stop(self):
        """停止Agent"""
        self._running = False
        logger.info(f"Agent {self.agent_id} stopped")


# 使用示例：创建具体的Agent实现
class DataAnalysisA2AAgent(BaseA2AAgent):
    """数据分析Agent的A2A实现"""
    
    def __init__(self, message_bus: A2AMessageBus):
        super().__init__(
            agent_id="data-agent-001",
            name="Data Analysis Agent",
            capabilities=[
                AgentCapability(
                    name="analyze",
                    description="Perform data analysis",
                    input_schema={"type": "object", "properties": {"data": {}}},
                    output_schema={"type": "object"}
                )
            ],
            message_bus=message_bus
        )
    
    async def execute_task(self, message: A2AMessage) -> Any:
        data = message.payload.get("data", {})
        logger.info(f"DataAgent analyzing data: {data}")
        
        # 模拟数据分析
        await asyncio.sleep(0.5)  # 模拟处理时间
        
        return {
            "summary": "Analysis complete",
            "record_count": 1000,
            "key_findings": ["Finding A", "Finding B"]
        }


async def demonstrate_a2a():
    """演示A2A协议的工作流程"""
    message_bus = A2AMessageBus()
    
    # 创建Agent
    data_agent = DataAnalysisA2AAgent(message_bus)
    
    # 启动消息总线和Agent
    await message_bus.start()
    await data_agent.start()
    
    # 模拟发送任务请求
    task_id = "task-123"
    print(f"Sending task request: {task_id}")
    
    await data_agent.send_task_request(
        receiver_id="data-agent-001",
        task_id=task_id,
        task_data={"data": {"source": "database", "query": "SELECT *"}}
    )
    
    # 等待处理
    await asyncio.sleep(1)
    
    # 检查任务结果
    result = data_agent.task_results.get(task_id)
    print(f"Task result: {result}")
    
    # 停止
    await data_agent.stop()
    await message_bus.stop()


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

# Python示例：客服多Agent系统的简化实现
class CustomerServiceMultiAgentSystem:
    """
    客服多Agent系统架构：
    - IntentAgent: 意图识别
    - ProductAgent: 商品推荐
    - OrderAgent: 订单查询
    - ReviewAgent: 结果审核
    """
    
    def __init__(self):
        self.intent_agent = IntentRecognitionAgent()
        self.product_agent = ProductRecommendationAgent()
        self.order_agent = OrderTrackingAgent()
        self.review_agent = QualityReviewAgent()
        self.orchestrator = TaskPlanner()
    
    async def handle_customer_message(self, user_id: str, message: str) -> str:
        # 步骤1：意图识别
        intent = await self.intent_agent.recognize(message)
        
        if intent == "product_inquiry":
            # 步骤2：商品推荐
            products = await self.product_agent.recommend(message)
            # 步骤3：结果审核
            response = await self.review_agent.review(products, context={"user_id": user_id})
            return response
        
        elif intent == "order_tracking":
            # 提取订单号
            order_id = self._extract_order_id(message)
            # 步骤2：订单查询
            order_info = await self.order_agent.track(order_id)
            # 步骤3：结果审核
            response = await self.review_agent.review(order_info, context={"user_id": user_id})
            return response
        
        else:
            return "抱歉，我暂时无法处理这类请求"

// Go示例：代码开发多Agent工作流
package main

import (
	"context"
	"fmt"
	"time"
)

// SDLCPhase 软件开发生命周期阶段
type SDLCPhase string

const (
	PhaseRequirements SDLCPhase = "requirements"
	PhaseDesign      SDLCPhase = "design"
	PhaseCode        SDLCPhase = "code"
	PhaseTest        SDLCPhase = "test"
	PhaseReview      SDLCPhase = "review"
	PhaseDeploy      SDLCPhase = "deploy"
)

// CodeTask 代码开发任务
type CodeTask struct {
	ID          string
	Description string
	Language    string
	Framework   string
	CurrentPhase SDLCPhase
	Requirements string
	Design      string
	Code        string
	Tests       string
	ReviewNotes string
}

// RequirementsAgent 需求分析Agent
type RequirementsAgent struct{}

func (a *RequirementsAgent) Analyze(task *CodeTask) (string, error) {
	fmt.Printf("[RequirementsAgent] Analyzing requirements for: %s\n", task.Description)
	time.Sleep(200 * time.Millisecond)
	
	task.Requirements = fmt.Sprintf("Functional Requirements:\n1. %s\n2. Error handling\n3. Logging\n\nNon-Functional Requirements:\n1. Performance\n2. Security\n3. Scalability", task.Description)
	
	return task.Requirements, nil
}

// DesignAgent 架构设计Agent
type DesignAgent struct{}

func (a *DesignAgent) Design(task *CodeTask) (string, error) {
	fmt.Printf("[DesignAgent] Designing architecture for: %s\n", task.Description)
	time.Sleep(300 * time.Millisecond)
	
	// 根据语言和框架生成架构设计
	task.Design = fmt.Sprintf(`Architecture Design for %s (%s/%s):

1. Project Structure:
   - cmd/: Entry points
   - internal/: Core business logic
   - pkg/: Reusable packages
   - api/: API definitions

2. Key Components:
   - Handler layer
   - Service layer
   - Repository layer

3. Data Models:
   - Define domain entities
   - Create database schemas

4. API Design:
   - RESTful endpoints
   - Request/Response formats`,
		task.Description, task.Language, task.Framework)
	
	return task.Design, nil
}

// CodeGenerationAgent 代码生成Agent
type CodeGenerationAgent struct{}

func (a *CodeGenerationAgent) Generate(task *CodeTask) (string, error) {
	fmt.Printf("[CodeGenerationAgent] Generating %s code for: %s\n", task.Language, task.Description)
	time.Sleep(500 * time.Millisecond)
	
	// 生成示例代码
	task.Code = fmt.Sprintf(`// main.go - Generated %s code
package main

import (
    "fmt"
    "log"
)

func main() {
    fmt.Println("Starting: %s")
    // TODO: Implement %s
}`, task.Description, task.Description)
	
	return task.Code, nil
}

// TestGenerationAgent 测试生成Agent
type TestGenerationAgent struct{}

func (a *TestGenerationAgent) GenerateTests(task *CodeTask) (string, error) {
	fmt.Printf("[TestGenerationAgent] Generating tests for: %s\n", task.Description)
	time.Sleep(300 * time.Millisecond)
	
	task.Tests = fmt.Sprintf(`// main_test.go - Generated tests
package main

import "testing"

func TestMain(t *testing.T) {
    // Test case for: %s
    expected := true
    result := true // TODO: Implement actual test
    if result != expected {
        t.Errorf("Test failed: expected %v, got %v", expected, result)
    }
}`, task.Description)
	
	return task.Tests, nil
}

// CodeReviewAgent 代码审查Agent
type CodeReviewAgent struct{}

func (a *CodeReviewAgent) Review(task *CodeTask) (string, error) {
	fmt.Printf("[CodeReviewAgent] Reviewing code for: %s\n", task.Description)
	time.Sleep(200 * time.Millisecond)
	
	task.ReviewNotes = fmt.Sprintf(`Code Review Report:
- Code Quality: Good
- Security: No major issues found
- Performance: Acceptable
- Suggestions: Consider adding more error handling`)
	
	return task.ReviewNotes, nil
}

// CodeDevelopmentWorkflow 代码开发工作流
type CodeDevelopmentWorkflow struct {
	requirementsAgent *RequirementsAgent
	designAgent       *DesignAgent
	codeAgent         *CodeGenerationAgent
	testAgent         *TestGenerationAgent
	reviewAgent       *CodeReviewAgent
}

func NewCodeDevelopmentWorkflow() *CodeDevelopmentWorkflow {
	return &CodeDevelopmentWorkflow{
		requirementsAgent: &RequirementsAgent{},
		designAgent:       &DesignAgent{},
		codeAgent:         &CodeGenerationAgent{},
		testAgent:         &TestGenerationAgent{},
		reviewAgent:       &CodeReviewAgent{},
	}
}

func (w *CodeDevelopmentWorkflow) Execute(ctx context.Context, task *CodeTask) error {
	fmt.Printf("\n=== Starting Code Development Workflow ===\n")
	fmt.Printf("Task: %s\n\n", task.Description)
	
	// 阶段1：需求分析
	fmt.Println("--- Phase 1: Requirements Analysis ---")
	if _, err := w.requirementsAgent.Analyze(task); err != nil {
		return fmt.Errorf("requirements analysis failed: %w", err)
	}
	fmt.Printf("Requirements:\n%s\n\n", task.Requirements)
	
	// 阶段2：架构设计
	fmt.Println("--- Phase 2: Architecture Design ---")
	if _, err := w.designAgent.Design(task); err != nil {
		return fmt.Errorf("design failed: %w", err)
	}
	fmt.Printf("Design:\n%s\n\n", task.Design)
	
	// 阶段3：代码生成
	fmt.Println("--- Phase 3: Code Generation ---")
	if _, err := w.codeAgent.Generate(task); err != nil {
		return fmt.Errorf("code generation failed: %w", err)
	}
	fmt.Printf("Generated Code:\n%s\n\n", task.Code)
	
	// 阶段4：测试生成
	fmt.Println("--- Phase 4: Test Generation ---")
	if _, err := w.testAgent.GenerateTests(task); err != nil {
		return fmt.Errorf("test generation failed: %w", err)
	}
	fmt.Printf("Generated Tests:\n%s\n\n", task.Tests)
	
	// 阶段5：代码审查
	fmt.Println("--- Phase 5: Code Review ---")
	if _, err := w.reviewAgent.Review(task); err != nil {
		return fmt.Errorf("code review failed: %w", err)
	}
	fmt.Printf("Review Notes:\n%s\n\n", task.ReviewNotes)
	
	fmt.Println("=== Code Development Workflow Complete ===")
	return nil
}

func main() {
	ctx := context.Background()
	
	// 创建工作流
	workflow := NewCodeDevelopmentWorkflow()
	
	// 创建任务
	task := &CodeTask{
		ID:          "task-001",
		Description: "User Authentication Module",
		Language:    "Go",
		Framework:   "Gin",
	}
	
	// 执行工作流
	if err := workflow.Execute(ctx, task); err != nil {
		fmt.Printf("Workflow failed: %v\n", err)
	}
}

# Python示例：多Agent系统的记忆系统实现
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from datetime import datetime
import numpy as np
from collections import deque

@dataclass
class MemoryItem:
    """记忆单元"""
    id: str
    content: str
    embedding: Optional[List[float]] = None
    created_at: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    last_accessed: datetime = field(default_factory=datetime.now)
    memory_type: str = "generic"  # short_term, long_term, episodic
    metadata: Dict[str, Any] = field(default_factory=dict)

class WorkingMemory:
    """短期记忆 - 固定大小的循环缓冲区"""
    
    def __init__(self, max_size: int = 100):
        self.max_size = max_size
        self.buffer: deque = deque(maxlen=max_size)
    
    def add(self, item: MemoryItem):
        self.buffer.append(item)
    
    def get_recent(self, n: int = 10) -> List[MemoryItem]:
        """获取最近n条记忆"""
        items = list(self.buffer)[-n:]
        for item in items:
            item.access_count += 1
            item.last_accessed = datetime.now()
        return items
    
    def clear(self):
        self.buffer.clear()
    
    def __len__(self):
        return len(self.buffer)


class VectorStore:
    """向量存储 - 用于长期记忆的语义检索"""
    
    def __init__(self):
        self.items: List[MemoryItem] = []
    
    def add(self, item: MemoryItem):
        """添加记忆项"""
        if item.embedding is None:
            # 在实际实现中，这里应该调用embedding模型
            item.embedding = self._generate_embedding(item.content)
        self.items.append(item)
    
    def search(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        """语义搜索"""
        query_embedding = self._generate_embedding(query)
        
        # 计算余弦相似度
        scores = []
        for item in self.items:
            if item.embedding:
                score = self._cosine_similarity(query_embedding, item.embedding)
                scores.append((score, item))
        
        # 排序并返回top_k
        scores.sort(key=lambda x: x[0], reverse=True)
        return [item for _, item in scores[:top_k]]
    
    def _generate_embedding(self, text: str) -> List[float]:
        """生成文本embedding - 简化实现"""
        # 实际实现应该使用sentence-transformers等库
        np.random.seed(hash(text) % (2**32))
        return np.random.randn(384).tolist()
    
    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        """计算余弦相似度"""
        a = np.array(a)
        b = np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
    
    def __len__(self):
        return len(self.items)


class EpisodicMemory:
    """情景记忆 - 记录完整的任务执行轨迹"""
    
    def __init__(self):
        self.episodes: Dict[str, List[Dict]] = {}  # task_id -> episode
    
    def start_episode(self, task_id: str, context: Dict[str, Any]):
        """开始一个新的情景"""
        self.episodes[task_id] = [{
            "type": "start",
            "timestamp": datetime.now().isoformat(),
            "context": context
        }]
    
    def record_step(self, task_id: str, step_type: str, data: Dict[str, Any]):
        """记录情景中的一个步骤"""
        if task_id not in self.episodes:
            self.episodes[task_id] = []
        
        self.episodes[task_id].append({
            "type": step_type,
            "timestamp": datetime.now().isoformat(),
            "data": data
        })
    
    def end_episode(self, task_id: str, outcome: str):
        """结束情景"""
        if task_id in self.episodes:
            self.episodes[task_id].append({
                "type": "end",
                "timestamp": datetime.now().isoformat(),
                "outcome": outcome
            })
    
    def get_episode(self, task_id: str) -> Optional[List[Dict]]:
        """获取完整的情景"""
        return self.episodes.get(task_id)
    
    def find_similar_episodes(self, current_task: str, top_k: int = 3) -> List[str]:
        """找到相似的历史情景"""
        # 简化实现：基于任务描述的简单匹配
        # 实际实现应该使用embedding相似度
        return list(self.episodes.keys())[-top_k:]


class MultiAgentMemory:
    """多Agent系统的统一记忆系统"""
    
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.working_memory = WorkingMemory(max_size=100)
        self.long_term_memory = VectorStore()
        self.episodic_memory = EpisodicMemory()
        self.current_task_id: Optional[str] = None
    
    def remember(self, content: str, memory_type: str = "generic", metadata: Optional[Dict] = None):
        """存储记忆"""
        item = MemoryItem(
            id=f"{self.agent_id}_{datetime.now().timestamp()}",
            content=content,
            memory_type=memory_type,
            metadata=metadata or {}
        )
        
        if memory_type == "short_term":
            self.working_memory.add(item)
        else:
            self.long_term_memory.add(item)
    
    def recall(self, query: str, search_type: str = "all") -> List[MemoryItem]:
        """检索记忆"""
        results = []
        
        if search_type in ["all", "recent"]:
            # 搜索短期记忆
            recent = self.working_memory.get_recent(10)
            results.extend([r for r in recent if query.lower() in r.content.lower()])
        
        if search_type in ["all", "semantic"]:
            # 语义搜索长期记忆
            semantic_results = self.long_term_memory.search(query)
            results.extend(semantic_results)
        
        return results
    
    def start_task(self, task_id: str, context: Dict[str, Any]):
        """开始新任务"""
        self.current_task_id = task_id
        self.episodic_memory.start_episode(task_id, context)
    
    def record_task_step(self, step_type: str, data: Dict[str, Any]):
        """记录任务步骤"""
        if self.current_task_id:
            self.episodic_memory.record_step(self.current_task_id, step_type, data)
    
    def complete_task(self, outcome: str):
        """完成任务"""
        if self.current_task_id:
            # 从工作记忆中提取关键信息存入长期记忆
            recent = self.working_memory.get_recent(5)
            for item in recent:
                item.memory_type = "long_term"
                self.long_term_memory.add(item)
            
            # 结束情景记录
            self.episodic_memory.end_episode(self.current_task_id, outcome)
            self.current_task_id = None
    
    def learn_from_past(self, current_task: str) -> List[str]:
        """从过去相似任务中学习"""
        similar_tasks = self.episodic_memory.find_similar_episodes(current_task)
        lessons = []
        
        for task_id in similar_tasks:
            episode = self.episodic_memory.get_episode(task_id)
            if episode:
                # 提取成功模式和失败教训
                for step in episode:
                    if step["type"] == "success" or step["type"] == "error":
                        lessons.append(step.get("data", {}).get("lesson", ""))
        
        return lessons


# 使用示例
def demonstrate_memory_system():
    # 创建Agent记忆系统
    agent_memory = MultiAgentMemory("data-agent-001")
    
    # 模拟任务执行
    agent_memory.start_task("task-001", {"description": "数据分析任务"})
    agent_memory.record_task_step("agent_call", {
        "agent": "search",
        "input": "查询销售数据",
        "lesson": "搜索关键词应该更精确"
    })
    
    # 添加记忆
    agent_memory.remember(
        content="数据清洗的关键是处理缺失值和异常值",
        memory_type="short_term",
        metadata={"topic": "data_processing"}
    )
    
    # 检索记忆
    results = agent_memory.recall("数据清洗")
    print(f"Retrieved {len(results)} memories about '数据清洗'")
    
    # 完成任务并学习
    agent_memory.complete_task("success")
    lessons = agent_memory.learn_from_past("数据分析")
    print(f"Learned {len(lessons)} lessons from past tasks")


if __name__ == "__main__":
    demonstrate_memory_system()

# Python示例：多Agent系统的容错机制
import asyncio
import time
from functools import wraps
from typing import Callable, Any, Optional, Type
from dataclasses import dataclass
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class RetryStrategy(Enum):
    EXPONENTIAL_BACKOFF = "exponential_backoff"
    LINEAR_BACKOFF = "linear_backoff"
    FIXED_DELAY = "fixed_delay"


@dataclass
class RetryConfig:
    max_attempts: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_BACKOFF
    retryable_exceptions: tuple = (ConnectionError, TimeoutError, asyncio.TimeoutError)


@dataclass
class CircuitBreakerState:
    failure_count: int = 0
    success_count: int = 0
    last_failure_time: Optional[float] = None
    state: str = "closed"  # closed, open, half_open


def with_retry(config: RetryConfig = None):
    """带重试装饰器"""
    if config is None:
        config = RetryConfig()
    
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(config.max_attempts):
                try:
                    return await func(*args, **kwargs)
                except config.retryable_exceptions as e:
                    last_exception = e
                    if attempt < config.max_attempts - 1:
                        delay = calculate_delay(attempt, config)
                        logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                        await asyncio.sleep(delay)
                    else:
                        logger.error(f"All {config.max_attempts} attempts failed")
            
            raise last_exception
        
        return wrapper
    return decorator


def calculate_delay(attempt: int, config: RetryConfig) -> float:
    """计算重试延迟"""
    if config.strategy == RetryStrategy.EXPONENTIAL_BACKOFF:
        delay = min(config.initial_delay * (2 ** attempt), config.max_delay)
    elif config.strategy == RetryStrategy.LINEAR_BACKOFF:
        delay = min(config.initial_delay * (attempt + 1), config.max_delay)
    else:  # FIXED_DELAY
        delay = config.initial_delay
    
    # 添加随机抖动
    import random
    delay *= (0.5 + random.random())
    return delay


class CircuitBreaker:
    """熔断器 - 防止级联故障"""
    
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0, recovery_ratio: float = 0.5):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.recovery_ratio = recovery_ratio
        self.state = CircuitBreakerState()
    
    @property
    def is_open(self) -> bool:
        if self.state.state == "open":
            # 检查是否超时
            if self.state.last_failure_time:
                elapsed = time.time() - self.state.last_failure_time
                if elapsed > self.recovery_timeout:
                    self.state.state = "half_open"
                    logger.info("Circuit breaker: transitioning to half_open")
            return self.state.state == "open"
        return False
    
    def record_success(self):
        """记录成功"""
        self.state.success_count += 1
        self.state.failure_count = 0
        
        if self.state.state == "half_open":
            if self.state.success_count >= 2:
                self.state.state = "closed"
                self.state.success_count = 0
                logger.info("Circuit breaker: recovered")
    
    def record_failure(self):
        """记录失败"""
        self.state.failure_count += 1
        self.state.last_failure_time = time.time()
        
        if self.state.failure_count >= self.failure_threshold:
            self.state.state = "open"
            logger.warning(f"Circuit breaker: opened after {self.state.failure_count} failures")


class AgentFaultTolerantClient:
    """Agent容错客户端"""
    
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.circuit_breaker = CircuitBreaker()
        self.retry_config = RetryConfig()
    
    async def execute_with_fallback(
        self, 
        primary_func: Callable,
        fallback_func: Optional[Callable] = None,
        *args, **kwargs
    ) -> Any:
        """使用降级的执行方法"""
        if self.circuit_breaker.is_open:
            logger.warning(f"Circuit breaker is open for {self.agent_id}, using fallback")
            if fallback_func:
                return await fallback_func(*args, **kwargs)
            raise Exception(f"Agent {self.agent_id} unavailable and no fallback provided")
        
        try:
            # 使用重试装饰器执行
            @with_retry(self.retry_config)
            async def execute():
                return await primary_func(*args, **kwargs)
            
            result = await execute()
            self.circuit_breaker.record_success()
            return result
            
        except Exception as e:
            self.circuit_breaker.record_failure()
            logger.error(f"Agent {self.agent_id} execution failed: {e}")
            
            if fallback_func:
                logger.info(f"Executing fallback for {self.agent_id}")
                return await fallback_func(*args, **kwargs)
            raise


class SandboxExecutor:
    """沙箱执行器 - 隔离Agent执行"""
    
    def __init__(self, timeout: float = 30.0, memory_limit: int = 512 * 1024 * 1024):
        self.timeout = timeout
        self.memory_limit = memory_limit
    
    async def execute(self, code: str, language: str) -> dict:
        """在沙箱中执行代码"""
        logger.info(f"Executing code in sandbox: language={language}")
        
        try:
            # 模拟执行
            await asyncio.sleep(0.5)
            
            return {
                "success": True,
                "output": f"Executed {language} code",
                "execution_time": 0.5
            }
            
        except asyncio.TimeoutError:
            return {
                "success": False,
                "error": "Execution timeout"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }


class TaskRollbackManager:
    """任务回滚管理器"""
    
    def __init__(self):
        self.checkpoints: dict = {}
    
    def create_checkpoint(self, task_id: str, state: dict):
        """创建检查点"""
        self.checkpoints[task_id] = {
            "timestamp": time.time(),
            "state": state.copy()
        }
        logger.info(f"Checkpoint created for task {task_id}")
    
    def rollback(self, task_id: str) -> Optional[dict]:
        """回滚到上一个检查点"""
        if task_id in self.checkpoints:
            checkpoint = self.checkpoints[task_id]
            logger.info(f"Rolling back task {task_id} to checkpoint at {checkpoint['timestamp']}")
            return checkpoint["state"]
        return None
    
    def clear_checkpoint(self, task_id: str):
        """清除检查点"""
        if task_id in self.checkpoints:
            del self.checkpoints[task_id]


# 使用示例
async def demonstrate_fault_tolerance():
    # 创建容错客户端
    client = AgentFaultTolerantClient("data-agent")
    sandbox = SandboxExecutor(timeout=10.0)
    rollback_manager = TaskRollbackManager()
    
    # 模拟执行
    async def primary_execute(data):
        # 可能失败的primary函数
        raise ConnectionError("Primary service unavailable")
    
    async def fallback_execute(data):
        # 降级方案
        logger.info("Executing fallback方案")
        return {"result": "from fallback", "data": data}
    
    # 带降级的执行
    result = await client.execute_with_fallback(
        primary_execute,
        fallback_execute,
        {"test": "data"}
    )
    print(f"Result: {result}")
    
    # 创建检查点并回滚
    rollback_manager.create_checkpoint("task-001", {"step": 1, "result": "partial"})
    rollback_manager.create_checkpoint("task-001", {"step": 2, "result": "partial2"})
    
    # 模拟回滚
    state = rollback_manager.rollback("task-001")
    print(f"Rolled back state: {state}")


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