AutoGen多代理系统架构深度解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# 核心消息传递示例
async def core_message_passing_example():
    # 创建运行时
    runtime = SingleThreadedAgentRuntime()
    
    # 注册代理
    await runtime.register_factory(
        "CoreAgent", 
        lambda: CoreAgent("核心处理代理")
    )
    
    # 异步消息传递
    result = await runtime.send_message(
        message={"task": "process_data", "data": "sample"},
        recipient=AgentId("CoreAgent", "default")
    )

1
2
3
4
5
6
7
8
9

# LLM集成示例
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key="your_api_key",
    # 支持模型热切换
    fallback_models=["gpt-3.5-turbo", "claude-3-sonnet"]
)

1
2
3
4

@dataclass
class AgentId:
    type: str    # 代理类型，如 "ChatAgent", "ToolAgent"
    key: str     # 代理实例键，如 "user_123", "default"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class AgentLifecycleManager:
    async def initialize_agent(self, agent_config: AgentConfig) -> Agent:
        """智能体初始化流程"""
        # 1. 加载配置和验证
        config = self._validate_config(agent_config)
        
        # 2. 注册能力和工具
        capabilities = await self._register_capabilities(config.capabilities)
        
        # 3. 建立连接池和资源
        connections = await self._establish_connections(config.endpoints)
        
        # 4. 初始化状态存储
        state_store = await self._init_state_store(config.persistence)
        
        # 5. 创建代理实例
        agent = await self._create_agent_instance(config, capabilities, connections, state_store)
        
        return agent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

class AgentMonitor:
    async def monitor_agent_health(self, agent: Agent) -> HealthStatus:
        """智能体健康监控"""
        # 状态监测
        health_metrics = {
            'memory_usage': await self._check_memory_usage(agent),
            'response_time': await self._measure_response_time(agent),
            'error_rate': await self._calculate_error_rate(agent),
            'connection_status': await self._check_connections(agent)
        }
        
        # 资源调度优化
        if health_metrics['memory_usage'] > 0.8:
            await self._trigger_memory_cleanup(agent)
        
        # 异常处理和自愈
        if health_metrics['error_rate'] > 0.1:
            await self._initiate_error_recovery(agent)
        
        return HealthStatus(metrics=health_metrics)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class AgentShutdownManager:
    async def graceful_shutdown(self, agent: Agent) -> None:
        """优雅关闭智能体"""
        # 1. 停止接收新请求
        await agent.stop_accepting_requests()
        
        # 2. 完成正在处理的任务
        await agent.wait_for_pending_tasks(timeout=30.0)
        
        # 3. 保存状态到持久化存储
        await self._save_agent_state(agent)
        
        # 4. 释放资源和连接
        await self._release_resources(agent)
        
        # 5. 清理内存和缓存
        await self._cleanup_memory(agent)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

{
  "specversion": "1.0",
  "id": "uuid-1234-5678",
  "source": "agent://ChatAgent/user_123", 
  "type": "com.microsoft.autogen.message.text",
  "datacontenttype": "application/json",
  "time": "2025-09-13T10:30:00Z",
  "data": {
    "content": "Hello, world!",
    "metadata": {}
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

service AgentRpc {
    // 双向流式通信通道
    rpc OpenChannel (stream Message) returns (stream Message);
    
    // 控制通道，用于状态管理
    rpc OpenControlChannel (stream ControlMessage) returns (stream ControlMessage);
    
    // 代理类型注册
    rpc RegisterAgent(RegisterAgentTypeRequest) returns (RegisterAgentTypeResponse);
    
    // 订阅管理
    rpc AddSubscription(AddSubscriptionRequest) returns (AddSubscriptionResponse);
    rpc RemoveSubscription(RemoveSubscriptionRequest) returns (RemoveSubscriptionResponse);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Python异步代理实现
class AsyncChatAgent(RoutedAgent):
    @message_handler
    async def handle_message(self, message: TextMessage, ctx: MessageContext) -> Response:
        # 异步处理消息
        result = await self.process_async(message.content)
        return Response(content=result)
        
    @event  
    async def handle_event(self, event: NotificationEvent, ctx: MessageContext) -> None:
        # 异步处理事件，无返回值
        await self.notify_async(event.data)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// .NET异步代理实现  
public class AsyncChatAgent : BaseAgent
{
    [MessageHandler]
    public async ValueTask<Response> HandleMessageAsync(TextMessage message, MessageContext context)
    {
        // 异步处理消息
        var result = await ProcessAsync(message.Content);
        return new Response { Content = result };
    }
}

1
2
3
4
5

FROM mcr.microsoft.com/dotnet/aspnet:8.0
WORKDIR /app
COPY . .
EXPOSE 5000
ENTRYPOINT ["dotnet", "AutoGen.Host.dll"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

apiVersion: apps/v1
kind: Deployment
metadata:
  name: autogen-gateway
spec:
  replicas: 3
  selector:
    matchLabels:
      app: autogen-gateway
  template:
    metadata:
      labels:
        app: autogen-gateway
    spec:
      containers:
      - name: gateway
        image: autogen/gateway:latest
        ports:
        - containerPort: 5000
        env:
        - name: REDIS_URL
          value: "redis://redis-service:6379"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

var builder = DistributedApplication.CreateBuilder(args);

// 添加Redis缓存
var redis = builder.AddRedis("cache");

// 添加Gateway服务
builder.AddProject<Projects.AutoGen_Gateway>("gateway")
       .WithReference(redis);

// 添加Worker服务  
builder.AddProject<Projects.AutoGen_Worker>("worker")
       .WithReference(redis)
       .WithReplicas(3);

builder.Build().Run();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

try:
    result = await agent_runtime.send_message(message, recipient)
except CantHandleException:
    # 代理无法处理该消息
    logger.warning(f"Agent {recipient} cannot handle message type {type(message)}")
    return ErrorResponse("Message type not supported")
except UndeliverableException:
    # 消息无法投递
    logger.error(f"Failed to deliver message to {recipient}")
    return ErrorResponse("Message delivery failed")  
except TimeoutException:
    # 处理超时
    logger.error(f"Message processing timeout for {recipient}")
    return ErrorResponse("Processing timeout")