AutoGen Python核心模块源码深度解析

# 上下文感知依赖注入
class AgentInstantiationContext:
    """上下文感知的依赖注入实现"""
    _context_stack: ContextVar[List[InstantiationContext]] = ContextVar('agent_context', default=[])
    
    @classmethod
    def populate_context(cls, agent_id: AgentId, runtime: AgentRuntime):
        """避免显式依赖配置，代理可以感知创建环境"""
        return _PopulateContext(agent_id, runtime)

# 双层路由设计
# 第一层：type(message)字典查找 - O(1)快速筛选
# 第二层：handler.router()条件匹配 - O(n)精确路由
# 在大多数场景下实现O(1)性能，同时保持路由灵活性

graph TB
    subgraph "消息发送路径 (热路径)"
        A1[BaseAgent.send_message] --> A2[AgentRuntime.send_message]
        A2 --> A3[MessageQueue.put]
        A3 --> A4[_process_next]
        A4 --> A5[_process_send]
        A5 --> A6[_get_agent]
        A6 --> A7[Agent.on_message]
        A7 --> A8[RoutedAgent.on_message_impl]
        A8 --> A9[handler.router]
        A9 --> A10[handler]
    end
    
    subgraph "代理创建路径 (冷路径)"
        B1[_get_agent] --> B2[AgentInstantiationContext.populate_context]
        B2 --> B3[factory_func]
        B3 --> B4[Agent.__init__]
        B4 --> B5[bind_id_and_runtime]
        B5 --> B6[cache_agent_instance]
    end
    
    subgraph "工具调用路径"
        C1[Workbench.call_tool] --> C2[FunctionTool.run_json]
        C2 --> C3[FunctionTool.run]
        C3 --> C4[prepare_kwargs]
        C4 --> C5[async/sync_execution]
        C5 --> C6[format_result]
    end
    
    subgraph "订阅匹配路径"
        D1[publish_message] --> D2[get_subscribed_recipients]
        D2 --> D3[subscription.is_match]
        D3 --> D4[subscription.map_to_agent]
        D4 --> D5[parallel_agent_processing]
    end
    
    style A6 fill:#e1f5fe
    style B3 fill:#f3e5f5
    style C3 fill:#e8f5e8
    style D3 fill:#fff3e0

graph TB
    subgraph "autogen-core 核心模块架构"
        subgraph "代理抽象层"
            A[Agent - 代理协议]
            BA[BaseAgent - 基础代理]
            RA[RoutedAgent - 路由代理]
            CA[ClosureAgent - 闭包代理]
        end
        
        subgraph "运行时层"
            AR[AgentRuntime - 运行时协议]
            STAR[SingleThreadedAgentRuntime - 单线程运行时]
            RC[RunContext - 运行上下文]
        end
        
        subgraph "消息处理层"
            MC[MessageContext - 消息上下文]
            MH[MessageHandler - 消息处理器]
            MHC[MessageHandlerContext - 处理器上下文]
        end
        
        subgraph "订阅与路由"
            S[Subscription - 订阅协议]
            TS[TypeSubscription - 类型订阅]
            TPS[TypePrefixSubscription - 前缀订阅]
            DS[DefaultSubscription - 默认订阅]
        end
        
        subgraph "标识与主题"
            AI[AgentId - 代理标识]
            AT[AgentType - 代理类型]
            TI[TopicId - 主题标识]
        end
        
        subgraph "序列化与通信"
            MS[MessageSerializer - 消息序列化器]
            SR[SerializationRegistry - 序列化注册表]
            UP[UnknownPayload - 未知载荷]
        end
        
        subgraph "工具与组件"
            T[Tool - 工具抽象]
            FT[FunctionTool - 函数工具]
            CC[ComponentConfig - 组件配置]
        end
    end
    
    %% 依赖关系
    BA --> A
    RA --> BA
    CA --> BA
    
    STAR --> AR
    RC --> STAR
    
    MH --> MC
    MHC --> MH
    
    TS --> S
    TPS --> S
    DS --> S
    
    AI --> AT
    TI --> AI
    
    MS --> SR
    UP --> MS
    
    FT --> T
    CC --> FT
    
    style A fill:#e1f5fe
    style STAR fill:#f3e5f5
    style S fill:#e8f5e8
    style MS fill:#fff3e0

# 代理在调用前只存在类型注册信息，没有实际实例
# 直到首次消息到达才创建实例

class LazyAgentSystem:
    """懒加载代理系统"""
    
    def __init__(self):
        self.agent_factories = {}     # 存储类型和工厂函数
        self.active_agents = {}       # 实际运行的实例
    
    async def register_agent_type(self, agent_type: str, factory: Callable):
        """注册代理类型"""
        self.agent_factories[agent_type] = factory
    
    async def get_or_create_agent(self, agent_id: AgentId):
        """按需创建代理实例"""
        if agent_id not in self.active_agents:
            factory = self.agent_factories[agent_id.type]
            self.active_agents[agent_id] = await factory()
        return self.active_agents[agent_id]

class MessageProcessingPattern:
    """消息处理模式分析"""
    
    def __init__(self):
        # 单一消息类型：顺畅流动，路由开销小
        # 多消息类型混合：需要额外路由开销
        # 路由复杂度影响消息流动效率
        pass
    
    def calculate_flow_efficiency(self, message_types: int, router_complexity: float) -> float:
        """计算消息流动效率"""
        return 1.0 / (1.0 + router_complexity * math.log(message_types + 1))

class ComponentConfigurationPattern:
    """组件配置模式"""
    
    # ComponentModel配置 -> _from_config()方法 -> 组件实例化
    # 依赖注入和生命周期管理
    
    def create_component(self, config: ComponentModel) -> Component:
        """组件创建过程"""
        # 解析配置
        genetic_code = self.parse_config(config)
        
        # 创建实例
        component_instance = self.instantiate_component(genetic_code)
        
        # 环境适应
        self.adapt_to_environment(component_instance)
        
        return component_instance

@runtime_checkable  
class Agent(Protocol):
    """代理协议定义"""
    async def on_message(self, message: Any, ctx: MessageContext) -> Any:
        """处理消息的核心方法"""
        ...

@runtime_checkable
class AgentRuntime(Protocol):
    """运行时协议定义"""
    async def send_message(self, message: Any, recipient: AgentId, ...) -> Any:
        """发送消息到指定代理"""
        ...
    
    async def publish_message(self, message: Any, topic_id: TopicId, ...) -> None:
        """发布消息到主题"""
        ...

class ChatAgent(RoutedAgent):
    @message_handler  # 通用消息处理装饰器
    async def handle_text_message(self, message: TextMessage, ctx: MessageContext) -> str:
        return f"收到消息: {message.content}"
    
    @event  # 事件处理装饰器  
    async def handle_notification(self, event: NotificationEvent, ctx: MessageContext) -> None:
        print(f"收到事件: {event.type}")
    
    @rpc  # RPC调用装饰器
    async def process_request(self, request: ProcessRequest, ctx: MessageContext) -> ProcessResponse:
        result = await self.complex_processing(request.data)
        return ProcessResponse(result=result)

class AgentId:
    """代理ID由类型和键组成，形成唯一标识"""
    
    def __init__(self, type: str | AgentType, key: str) -> None:
        if isinstance(type, AgentType):
            type = type.type
            
        # 验证类型格式：只允许字母、数字、下划线、连字符和点
        if not is_valid_agent_type(type):
            raise ValueError(f"Invalid agent type: {type}")
            
        self._type = type  # 代理类型，如 "ChatAgent"
        self._key = key    # 代理实例键，如 "user_123"

def is_valid_agent_type(value: str) -> bool:
    """验证代理类型格式
    
    允许的字符：字母、数字、下划线(_)、连字符(-)、点(.)
    正则表达式：^[\w\-\.]+\Z
    """
    return bool(re.match(r"^[\w\-\.]+\Z", value))

class AgentId:
    def __hash__(self) -> int:
        """支持作为字典键"""
        return hash((self._type, self._key))
    
    def __str__(self) -> str:
        """字符串表示：type/key"""
        return f"{self._type}/{self._key}"
    
    @classmethod  
    def from_str(cls, agent_id: str) -> Self:
        """从字符串解析AgentId：'ChatAgent/user_123'"""
        items = agent_id.split("/", maxsplit=1)
        if len(items) != 2:
            raise ValueError(f"Invalid agent id: {agent_id}")
        type, key = items[0], items[1]
        return cls(type, key)

class EventTypeAnalysis:
    """事件类型分析"""
    
    def __init__(self, topic_id: TopicId):
        # 解析事件类型的层次结构
        self.type_hierarchy = {
            'domain': topic_id.type.split('.')[0],       # 域（如com）  
            'org': topic_id.type.split('.')[1],          # 组织（如microsoft）
            'service': topic_id.type.split('.')[2],      # 服务（如autogen）
            'category': topic_id.type.split('.')[3],     # 分类（如chat）
            'event': topic_id.type.split('.')[-1],       # 事件（如message）
            'source': topic_id.source                    # 来源（发送者信息）
        }
    
    def calculate_type_similarity(self, other_topic: TopicId) -> float:
        """计算事件类型相似度"""
        common_segments = self._count_common_segments(other_topic)
        total_segments = self._count_total_segments()
        return common_segments / total_segments

@dataclass(eq=True, frozen=True)
class TopicId:
    """主题标识，兼容CloudEvents规范"""
    
    type: str    # 事件类型，如 "com.microsoft.autogen.chat.message"
    source: str  # 事件源，如 "agent://ChatAgent/user_123"
    
    def __post_init__(self) -> None:
        """验证主题类型格式"""
        if not is_valid_topic_type(self.type):
            raise ValueError(f"Invalid topic type: {self.type}")

def is_valid_topic_type(value: str) -> bool:
    """验证主题类型格式
    
    允许的字符：字母、数字、下划线(_)、连字符(-)、点(.)、冒号(:)、等号(=)
    正则表达式：^[\w\-\.\:\=]+\Z
    """
    return bool(re.match(r"^[\w\-\.\:\=]+\Z", value))

# 系统级事件
SYSTEM_STARTUP = TopicId("com.microsoft.autogen.system.startup", "runtime://system")
SYSTEM_SHUTDOWN = TopicId("com.microsoft.autogen.system.shutdown", "runtime://system")

# 代理生命周期事件  
AGENT_CREATED = TopicId("com.microsoft.autogen.agent.created", "agent://ChatAgent/user_123")
AGENT_DESTROYED = TopicId("com.microsoft.autogen.agent.destroyed", "agent://ChatAgent/user_123")

# 业务事件
CHAT_MESSAGE = TopicId("com.example.chat.message", "agent://ChatAgent/user_123")
TASK_COMPLETED = TopicId("com.example.task.completed", "agent://TaskAgent/task_456")

@dataclass
class MessageContext:
    """消息处理上下文"""
    sender: AgentId | None          # 发送方代理ID
    topic_id: TopicId | None        # 主题ID（发布消息时使用）  
    is_rpc: bool                    # 是否为RPC调用
    cancellation_token: CancellationToken  # 取消令牌
    message_id: str                 # 消息唯一标识

@message_handler装饰器 → 
  get_type_hints(func) → 
    get_types(type_hints["message"]) → 
      create_wrapper_function() → 
        add_handler_metadata() → 
          wrapper.target_types = [MessageType] → 
            RoutedAgent._discover_handlers() → 
              _handlers[MessageType].append(handler)

运行时路由调用链：
RoutedAgent.on_message_impl() → 
  _handlers.get(type(message)) → 
    handler.router(message, ctx) → 
      handler(self, message, ctx) → 
        func(self, message, ctx)

@rpc装饰器路由：
handler.router = lambda msg, ctx: ctx.is_rpc and match(msg, ctx)

@event装饰器路由：  
handler.router = lambda msg, ctx: (not ctx.is_rpc) and match(msg, ctx)

def message_handler(
    func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]] = None,
    *,
    strict: bool = True,  # 严格类型检查
    match: None | Callable[[ReceivesT, MessageContext], bool] = None,  # 二次路由匹配
) -> MessageHandler[AgentT, ReceivesT, ProducesT]:
    """消息处理器装饰器
    
    Args:
        func: 被装饰的异步方法
        strict: 启用严格类型检查
        match: 二次路由匹配函数
    """
    
    def decorator(func: Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]]) -> MessageHandler[AgentT, ReceivesT, ProducesT]:
        # 提取类型信息
        type_hints = get_type_hints(func)
        target_types = get_types(type_hints["message"])  # 输入类型
        return_types = get_types(type_hints["return"])   # 返回类型
        
        @wraps(func)
        async def wrapper(self: AgentT, message: ReceivesT, ctx: MessageContext) -> ProducesT:
            # 类型检查
            if type(message) not in target_types:
                if strict:
                    raise CantHandleException(f"消息类型 {type(message)} 不在目标类型 {target_types} 中")
                else:
                    logger.warning(f"消息类型不匹配: {type(message)} not in {target_types}")
            
            # 执行处理逻辑
            return_value = await func(self, message, ctx)
            
            # 返回类型检查
            if return_value is not None and type(return_value) not in return_types:
                if strict:
                    raise RuntimeError(f"返回类型 {type(return_value)} 不在预期类型 {return_types} 中")
                else:
                    logger.warning(f"返回类型不匹配: {type(return_value)} not in {return_types}")
            
            return return_value
        
        # 添加元数据
        wrapper.target_types = target_types
        wrapper.produces_types = return_types
        wrapper.is_message_handler = True
        wrapper.router = match or (lambda message, ctx: True)
        
        return cast(MessageHandler[AgentT, ReceivesT, ProducesT], wrapper)
    
    return decorator if func is None else decorator(func)

def event(func: Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, None]]) -> MessageHandler[AgentT, ReceivesT, None]:
    """事件处理装饰器 - 无返回值"""
    return message_handler(func)

def rpc(
    func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]] = None,
    *,
    strict: bool = True,
    match: None | Callable[[ReceivesT, MessageContext], bool] = None,
) -> MessageHandler[AgentT, ReceivesT, ProducesT]:
    """RPC调用装饰器 - 必须有返回值"""
    
    def decorator(func: Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]]) -> MessageHandler[AgentT, ReceivesT, ProducesT]:
        handler = message_handler(func, strict=strict, match=match)
        
        # 验证RPC方法必须有返回值
        type_hints = get_type_hints(func)
        return_type = type_hints.get("return")
        if return_type is None or return_type is type(None):
            raise AssertionError("RPC方法必须有返回值")
        
        return handler
    
    return decorator if func is None else decorator(func)

@runtime_checkable
class Subscription(Protocol):
    """订阅协议定义"""
    
    @property
    def id(self) -> str:
        """订阅唯一标识"""
        ...
    
    def is_match(self, topic_id: TopicId) -> bool:
        """检查主题ID是否匹配此订阅"""
        ...
    
    def map_to_agent(self, topic_id: TopicId) -> AgentId:
        """将主题ID映射到代理ID"""
        ...

SubscriptionManager.get_subscribed_recipients() → 
  subscription.is_match(topic_id) → 
    TypeSubscription.is_match() → [topic_id.type == self.topic_type] → 
      subscription.map_to_agent(topic_id) → 
        AgentId(self.agent_type, "default")

TypePrefixSubscription.is_match() → [topic_id.type.startswith(prefix)] → 
  subscription.map_to_agent(topic_id) → 
    parse_agent_source(topic_id.source) → 
      AgentId(self.agent_type, parsed_key)

@dataclass
class TypeSubscription:
    """类型订阅 - 精确匹配主题类型"""
    
    id: str
    topic_type: str      # 要匹配的主题类型
    agent_type: str      # 目标代理类型
    
    def is_match(self, topic_id: TopicId) -> bool:
        """精确匹配主题类型"""
        return topic_id.type == self.topic_type
    
    def map_to_agent(self, topic_id: TopicId) -> AgentId:
        """映射到指定代理类型的默认实例"""
        if not self.is_match(topic_id):
            raise CantHandleException(f"主题 {topic_id} 不匹配订阅 {self.topic_type}")
        return AgentId(self.agent_type, "default")

# 使用示例
chat_subscription = TypeSubscription(
    id="chat-subscription-001",
    topic_type="com.example.chat.message", 
    agent_type="ChatAgent"
)

@dataclass  
class TypePrefixSubscription:
    """前缀订阅 - 匹配主题类型前缀"""
    
    id: str
    topic_type_prefix: str  # 主题类型前缀
    agent_type: str         # 目标代理类型
    
    def is_match(self, topic_id: TopicId) -> bool:
        """前缀匹配主题类型"""
        return topic_id.type.startswith(self.topic_type_prefix)
    
    def map_to_agent(self, topic_id: TopicId) -> AgentId:
        """映射到指定代理类型，使用源作为键"""
        if not self.is_match(topic_id):
            raise CantHandleException(f"主题 {topic_id} 不匹配前缀 {self.topic_type_prefix}")
        
        # 从主题源中提取代理键
        # 例如：agent://ChatAgent/user_123 -> user_123
        if topic_id.source.startswith("agent://"):
            agent_path = topic_id.source[8:]  # 移除 "agent://" 前缀
            if "/" in agent_path:
                _, key = agent_path.split("/", 1)
                return AgentId(self.agent_type, key)
        
        return AgentId(self.agent_type, "default")

# 使用示例
system_subscription = TypePrefixSubscription(
    id="system-subscription-001",
    topic_type_prefix="com.microsoft.autogen.system.",
    agent_type="SystemAgent" 
)

class SingleThreadedAgentRuntime(AgentRuntime):
    """单线程代理运行时 - 基于实际源码实现
    
    特点：
    - 使用单个asyncio队列处理所有消息
    - 消息按接收顺序处理，每个消息在独立的asyncio任务中并发处理
    - 支持点对点消息和发布订阅两种通信模式
    - 内置干预处理器和链路追踪支持
    - 适用于开发和独立应用，不适合高吞吐量场景
    """
    
    def __init__(
        self,
        *,
        intervention_handlers: List[InterventionHandler] | None = None,
        tracer_provider: TracerProvider | None = None,
        ignore_unhandled_exceptions: bool = True,
    ) -> None:
        # 核心消息队列 - 处理三种消息信封类型
        self._message_queue: Queue[
            PublishMessageEnvelope | SendMessageEnvelope | ResponseMessageEnvelope
        ] = Queue()
        
        # 代理工厂管理 - 支持懒加载和依赖注入
        self._agent_factories: Dict[
            str, Callable[[], Agent | Awaitable[Agent]] | Callable[[AgentRuntime, AgentId], Agent | Awaitable[Agent]]
        ] = {}
        
        # 实例化代理缓存
        self._instantiated_agents: Dict[AgentId, Agent] = {}
        
        # 中间件和干预处理
        self._intervention_handlers = intervention_handlers
        self._background_tasks: Set[Task[Any]] = set()
        
        # 订阅管理器 - 处理发布订阅路由
        self._subscription_manager = SubscriptionManager()
        
        # 序列化注册表 - 管理消息类型序列化器
        self._serialization_registry = SerializationRegistry()
        
        # 运行控制和异常处理
        self._ignore_unhandled_handler_exceptions = ignore_unhandled_exceptions
        self._background_exception: BaseException | None = None
        self._run_context: RunContext | None = None
        
        # 代理类型验证
        self._agent_instance_types: Dict[str, Type[Agent]] = {}

BaseAgent.send_message() → 
  AgentRuntime.send_message() → 
    MessageQueue.put(SendMessageEnvelope) → 
      _process_next() → 
        _process_send() → 
          _get_agent() → 
            Agent.on_message() → 
              RoutedAgent.on_message_impl() → 
                handlers.get(type(message)) → 
                  handler.router(message, ctx) → 
                    handler(self, message, ctx)

BaseAgent.publish_message() → 
  AgentRuntime.publish_message() → 
    MessageQueue.put(PublishMessageEnvelope) → 
      _process_next() → 
        _process_publish() → 
          get_subscribed_recipients() → 
            subscription.is_match(topic_id) → 
              subscription.map_to_agent(topic_id) → 
                Agent.on_message() → [并行处理多个订阅者]

AgentRuntime._get_agent() → 
  _agent_factories.get(agent_type) → 
    AgentInstantiationContext.populate_context() → 
      factory_func() → 
        Agent.__init__() → 
          AgentInstantiationContext.current_runtime() → 
            bind_id_and_runtime() → 
              _instantiated_agents[agent_id] = agent

async def _process_next(self) -> None:
    """处理队列中的下一个消息 - 基于实际源码实现"""
    
    # 检查后台异常
    if self._background_exception is not None:
        e = self._background_exception
        self._background_exception = None
        self._message_queue.shutdown(immediate=True)
        raise e

    try:
        # 从队列获取消息信封
        message_envelope = await self._message_queue.get()
    except QueueShutDown:
        if self._background_exception is not None:
            e = self._background_exception
            self._background_exception = None
            raise e from None
        return

    # 基于消息信封类型进行模式匹配处理
    match message_envelope:
        case SendMessageEnvelope(message=message, sender=sender, recipient=recipient, future=future):
            # 点对点消息处理
            await self._handle_send_message_with_intervention(message_envelope)
            
        case PublishMessageEnvelope(message=message, sender=sender, topic_id=topic_id):
            # 发布订阅消息处理
            await self._handle_publish_message_with_intervention(message_envelope)
            
        case ResponseMessageEnvelope(message=message, future=future):
            # 响应消息处理
            await self._handle_response_message(message_envelope)
        
        case _:
            logger.warning(f"未知消息信封类型: {type(message_envelope)}")
        self._message_queue.task_done()

async def _handle_send_message_with_intervention(self, message_envelope: SendMessageEnvelope) -> None:
    """处理点对点消息（包含干预处理）"""
    
    # 应用干预处理器
    if self._intervention_handlers is not None:
        for handler in self._intervention_handlers:
            try:
                message_context = MessageContext(
                    sender=message_envelope.sender,
                    topic_id=None,
                    is_rpc=True,
                    cancellation_token=message_envelope.cancellation_token,
                    message_id=message_envelope.message_id,
                )
                
                # 调用干预处理器
                processed_message = await handler.on_send(
                    message_envelope.message, 
                    message_context=message_context, 
                    recipient=message_envelope.recipient
                )
                
                # 检查消息是否被丢弃
                if processed_message is DropMessage or isinstance(processed_message, DropMessage):
                    logger.info(f"消息被干预处理器丢弃: {message_envelope.message_id}")
                    message_envelope.future.set_exception(MessageDroppedException())
                    return
                
                # 更新消息内容
                message_envelope.message = processed_message
                
            except BaseException as e:
                message_envelope.future.set_exception(e)
                return
    
    # 创建后台任务处理消息
    task = asyncio.create_task(self._process_send(message_envelope))
    self._background_tasks.add(task)
    task.add_done_callback(self._background_tasks.discard)

async def _process_send(self, message_envelope: SendMessageEnvelope) -> None:
    """处理点对点消息的核心逻辑 - 基于实际_process_send方法"""
    
    recipient = message_envelope.recipient
    
    # 验证代理类型是否存在
    if recipient.type not in self._known_agent_names:
        message_envelope.future.set_exception(LookupError(f"代理类型 '{recipient.type}' 不存在"))
        return
    
    try:
        logger.info(f"为 {recipient} 调用消息处理器，消息类型: {type(message_envelope.message).__name__}")
        
        # 获取或创建代理实例
        recipient_agent = await self._get_agent(recipient)
        
        # 构建消息上下文
        message_context = MessageContext(
            sender=message_envelope.sender,
            topic_id=None,
            is_rpc=True,
            cancellation_token=message_envelope.cancellation_token,
            message_id=message_envelope.message_id,
        )
        
        # 调用代理处理消息
        with MessageHandlerContext.populate_context(recipient_agent.id):
            response = await recipient_agent.on_message(
                message_envelope.message,
                ctx=message_context,
            )
        
        # 创建响应消息信封并加入队列
        await self._message_queue.put(
            ResponseMessageEnvelope(
                message=response,
                future=message_envelope.future,
                sender=message_envelope.recipient,
                recipient=message_envelope.sender,
                metadata=get_telemetry_envelope_metadata(),
            )
        )
        
    except CancelledError as e:
        if not message_envelope.future.cancelled():
            message_envelope.future.set_exception(e)
    except BaseException as e:
        message_envelope.future.set_exception(e)
    finally:
        self._message_queue.task_done()

async def _process_publish(self, message_envelope: PublishMessageEnvelope) -> None:
    """处理发布订阅消息 - 基于实际_process_publish方法"""
    
    try:
        responses: List[Awaitable[Any]] = []
        
        # 获取所有订阅接收者
        recipients = await self._subscription_manager.get_subscribed_recipients(message_envelope.topic_id)
        
        for agent_id in recipients:
            # 避免发送消息给发送者自己
            if message_envelope.sender is not None and agent_id == message_envelope.sender:
                continue

            logger.info(f"为 {agent_id.type} 调用消息处理器，发布消息类型: {type(message_envelope.message).__name__}")
            
            # 构建消息上下文
            message_context = MessageContext(
                sender=message_envelope.sender,
                topic_id=message_envelope.topic_id,
                is_rpc=False,
                cancellation_token=message_envelope.cancellation_token,
                message_id=message_envelope.message_id,
            )
            
            # 获取代理并异步处理消息
            agent = await self._get_agent(agent_id)
            
            async def _on_message(agent: Agent, message_context: MessageContext) -> Any:
                with MessageHandlerContext.populate_context(agent.id):
                    return await agent.on_message(message_envelope.message, message_context)
            
            # 创建并行处理任务
            response_task = asyncio.create_task(_on_message(agent, message_context))
            responses.append(response_task)
        
        # 等待所有订阅者处理完成
        if responses:
            await asyncio.gather(*responses, return_exceptions=True)
            
    except Exception as e:
        logger.error(f"处理发布消息时发生异常: {e}", exc_info=True)
        if not self._ignore_unhandled_handler_exceptions:
            raise
    finally:
        self._message_queue.task_done()

sequenceDiagram
    participant App as 应用程序
    participant RT as AgentRuntime
    participant AIC as AgentInstantiationContext
    participant AF as AgentFactory
    participant Agent as Agent实例
    participant SR as SerializationRegistry

    Note over App,SR: 代理注册和创建完整流程

    App->>RT: register_factory("ChatAgent", factory_func)
    activate RT
    
    RT->>RT: 验证代理类型格式
    RT->>RT: 存储到_agent_factories[type]
    RT->>App: 返回AgentType
    deactivate RT

    Note over App,SR: 代理按需创建流程 (懒加载)

    App->>RT: send_message(msg, AgentId("ChatAgent", "user123"))
    activate RT
    
    RT->>RT: _get_agent(agent_id)
    
    alt 代理未实例化
        RT->>RT: 检查_instantiated_agents[agent_id]
        RT->>AF: 从_agent_factories获取工厂函数
        
        Note over RT,AIC: AgentInstantiationContext依赖注入
        RT->>AIC: populate_context(agent_id, runtime)
        activate AIC
        
        RT->>AF: 调用factory_func()
        AF->>Agent: 创建代理实例
        activate Agent
        
        Agent->>AIC: AgentInstantiationContext.current_runtime()
        AIC->>Agent: 返回运行时实例
        Agent->>AIC: AgentInstantiationContext.current_agent_id()
        AIC->>Agent: 返回代理ID
        
        Agent->>Agent: 自动绑定_runtime和_id属性
        deactivate AIC
        
        AF->>RT: 返回代理实例
        
        Note over RT: 即时消息类型注册 (JIT)
        RT->>Agent: 调用_handles_types()获取消息类型
        Agent->>SR: 注册消息序列化器
        
        RT->>RT: _instantiated_agents[agent_id] = agent
        deactivate Agent
    else 代理已存在
        RT->>RT: 从_instantiated_agents返回缓存实例
    end
    
    RT->>Agent: on_message(message, context)
    activate Agent
    Agent->>Agent: 路由到对应的消息处理器
    Agent->>RT: 返回处理结果
    deactivate Agent
    
    RT->>App: 返回最终响应
    deactivate RT

# 基于实际 _get_agent 方法的实现
async def _get_agent(self, agent_id: AgentId) -> Agent:
    """获取或创建代理实例的核心方法"""
    
    # 1. 检查代理实例缓存
    if agent_id in self._instantiated_agents:
        return self._instantiated_agents[agent_id]
    
    # 2. 验证代理类型是否已注册
    if agent_id.type not in self._agent_factories:
        raise LookupError(f"代理类型 '{agent_id.type}' 未注册")
    
    # 3. 获取代理工厂函数
    factory = self._agent_factories[agent_id.type]
    
    # 4. 使用AgentInstantiationContext创建代理
    with AgentInstantiationContext.populate_context(agent_id, self):
        try:
            # 调用工厂函数创建代理实例
            if inspect.iscoroutinefunction(factory):
                if len(inspect.signature(factory).parameters) == 0:
                    agent = await factory()  # 无参数异步工厂
                else:
                    agent = await factory(self, agent_id)  # 带参数异步工厂
            else:
                if len(inspect.signature(factory).parameters) == 0:
                    agent = factory()  # 无参数同步工厂
                else:
                    agent = factory(self, agent_id)  # 带参数同步工厂
            
            # 5. 验证代理类型
            expected_type = self._agent_instance_types.get(agent_id.type)
            if expected_type is not None and not isinstance(agent, expected_type):
                raise TypeError(f"工厂返回的代理类型不匹配，期望: {expected_type}, 实际: {type(agent)}")
            
            # 6. 绑定代理ID和运行时（如果未自动绑定）
            if not hasattr(agent, '_id') or not hasattr(agent, '_runtime'):
                await agent.bind_id_and_runtime(agent_id, self)
            
            # 7. 缓存代理实例
            self._instantiated_agents[agent_id] = agent
            
            logger.debug(f"成功创建代理实例: {agent_id}")
            return agent
            
        except Exception as e:
            logger.error(f"创建代理实例失败: {agent_id}, 错误: {e}", exc_info=True)
            raise

# .NET版本的EnsureAgentAsync实现
async def ensure_agent_async(self, agent_id: AgentId) -> IHostableAgent:
    """
    .NET版本的代理确保方法 - 基于InProcessRuntime源码
    
    这个方法实现了代理的按需创建和即时消息类型注册，
    提供运行时性能和资源利用
    """
    
    # 1. 检查代理实例字典
    if not self.agentInstances.TryGetValue(agent_id, out agent):
        # 2. 从工厂字典获取工厂函数
        if not self.agentFactories.TryGetValue(agent_id.Type, out factory_func):
            raise Exception(f"Agent with name {agent_id.Type} not found.")
        
        # 3. 调用工厂函数创建代理实例
        agent = await factory_func(agent_id, this)
        
        # 4. 即时注册消息类型 (Just-in-Time Registration)
        # 这确保了代理能够正确序列化和反序列化它处理的消息类型
        agent.RegisterHandledMessageTypes(this.serializationRegistry)
        
        # 5. 缓存代理实例
        this.agentInstances.Add(agent_id, agent)
    
    return this.agentInstances[agent_id]

#### 消息路由机制时序图 (基于RoutedAgent源码)

```mermaid
sequenceDiagram
    participant Runtime as AgentRuntime
    participant Agent as RoutedAgent
    participant MH as MessageHandler
    participant Router as RouterFunction

    Note over Runtime,Router: RoutedAgent消息路由完整流程

    Runtime->>Agent: on_message(message, ctx)
    activate Agent
    
    Agent->>Agent: on_message_impl(message, ctx)
    Note over Agent: 获取消息类型
    Agent->>Agent: key_type = type(message)
    
    Agent->>Agent: 查找handlers = self._handlers.get(key_type)
    
    alt 找到匹配的处理器
        loop 遍历所有匹配的处理器
            Agent->>Router: handler.router(message, ctx)
            activate Router
            Router->>Router: 执行匹配函数
            Router->>Agent: 返回匹配结果
            deactivate Router
            
            alt 匹配成功
                Agent->>MH: handler(self, message, ctx)
                activate MH
                MH->>MH: 执行业务逻辑
                MH->>Agent: 返回处理结果
                deactivate MH
                Agent->>Runtime: 返回结果
                Note over Agent: 使用第一个匹配的处理器，跳过其余
                deactivate Agent
            else 匹配失败
                Note over Agent: 继续尝试下一个处理器
            end
        end
        
        Agent->>Agent: on_unhandled_message(message, ctx)
        Note over Agent: 所有处理器都不匹配
        Agent->>Runtime: 默认处理或抛出CantHandleException
        deactivate Agent
    else 未找到处理器
        Agent->>Agent: on_unhandled_message(message, ctx)
        Note over Agent: 记录未处理消息
        Agent->>Runtime: 抛出CantHandleException
        deactivate Agent
    end

AgentRuntime.save_state() → 
  [for agent_id in _instantiated_agents] → 
    _get_agent(agent_id) → 
      agent.save_state() → 
        [BaseAgent默认] warnings.warn("save_state not implemented") → 
        [自定义实现] serialize_agent_state() → 
          return state_dict → 
            runtime_state[str(agent_id)] = agent_state

AgentRuntime.load_state() → 
  [for agent_id_str in state] → 
    AgentId.from_str(agent_id_str) → 
      _get_agent(agent_id) → 
        agent.load_state(state[agent_id_str]) → 
          [BaseAgent默认] warnings.warn("load_state not implemented") → 
          [自定义实现] deserialize_and_restore_state() → 
            update_agent_internal_state()

BaseAgent.register() → 
  cls._handles_types() → 
    cls._discover_handlers() → 
      [for handler in handlers] → 
        handler.target_types → 
          try_get_known_serializers_for_type(type) → 
            SerializationRegistry.get_serializers() → 
              runtime.add_message_serializer(serializer) → 
                _serialization_registry.register(type, serializer)

sequenceDiagram
    participant App as 应用程序
    participant RT as AgentRuntime
    participant SIC as SubscriptionInstantiationContext
    participant SM as SubscriptionManager
    participant SR as SerializationRegistry
    participant AgentClass as 代理类

    Note over App,AgentClass: BaseAgent.register()完整注册流程

    App->>RT: BaseAgent.register(runtime, "ChatAgent", factory_func)
    activate RT
    
    Note over RT: 步骤1: 注册代理工厂
    RT->>RT: register_factory(AgentType("ChatAgent"), factory_func)
    RT->>RT: 存储到_agent_factories["ChatAgent"]
    
    Note over RT: 步骤2: 处理类级订阅
    alt skip_class_subscriptions=False
        RT->>SIC: populate_context(AgentType("ChatAgent"))
        activate SIC
        
        RT->>AgentClass: 调用cls._unbound_subscriptions()
        AgentClass->>RT: 返回订阅工厂函数列表
        
        loop 对每个unbound_subscription
            RT->>RT: 调用subscription_factory()
            
            alt 返回异步结果
                RT->>RT: await subscription_factory()
            else 返回同步结果
                RT->>RT: 直接使用结果
            end
            
            RT->>SM: add_subscription(subscription)
            SM->>SM: 添加到订阅映射表
        end
        
        deactivate SIC
    end
    
    Note over RT: 步骤3: 添加直接消息订阅
    alt skip_direct_message_subscription=False
        RT->>SM: add_subscription(TypePrefixSubscription("ChatAgent:", "ChatAgent"))
        SM->>SM: 添加前缀订阅规则
    end
    
    Note over RT: 步骤4: 注册消息序列化器
    RT->>AgentClass: 调用cls._handles_types()
    AgentClass->>RT: 返回(message_type, serializer)元组列表
    
    loop 对每个消息类型
        RT->>SR: add_message_serializer(serializer)
        SR->>SR: 注册序列化器
    end
    
    RT->>App: 返回AgentType("ChatAgent")
    deactivate RT

@classmethod
async def register(
    cls,
    runtime: AgentRuntime,
    type: str,
    factory: Callable[[], Self | Awaitable[Self]],
    *,
    skip_class_subscriptions: bool = False,
    skip_direct_message_subscription: bool = False,
) -> AgentType:
    """
    代理注册的完整实现 - 基于实际BaseAgent.register源码
    
    这个方法实现了代理注册的完整生命周期：
    1. 工厂函数注册 - 支持懒加载代理创建
    2. 类级订阅处理 - 处理@default_subscription等装饰器定义的订阅
    3. 直接消息订阅 - 添加基于代理类型的前缀订阅
    4. 消息序列化器注册 - JIT注册代理处理的消息类型
    """
    
    # 1. 注册代理工厂到运行时
    agent_type = AgentType(type)
    agent_type = await runtime.register_factory(
        type=agent_type, 
        agent_factory=factory, 
        expected_class=cls
    )
    
    # 2. 处理类级订阅 (如果不跳过)
    if not skip_class_subscriptions:
        with SubscriptionInstantiationContext.populate_context(agent_type):
            subscriptions: List[Subscription] = []
            
            # 获取类定义的未绑定订阅
            for unbound_subscription in cls._unbound_subscriptions():
                subscriptions_list_result = unbound_subscription()
                
                # 处理异步订阅工厂
                if inspect.isawaitable(subscriptions_list_result):
                    subscriptions_list = await subscriptions_list_result
                else:
                    subscriptions_list = subscriptions_list_result

                subscriptions.extend(subscriptions_list)
        
        # 将所有订阅注册到运行时
        for subscription in subscriptions:
            await runtime.add_subscription(subscription)

    # 3. 添加直接消息订阅 (TypePrefixSubscription)
    if not skip_direct_message_subscription:
        try:
        await runtime.add_subscription(
            TypePrefixSubscription(
                    # 前缀必须包含":"以避免与其他代理类型冲突
                topic_type_prefix=agent_type.type + ":",
                agent_type=agent_type.type,
            )
        )
        except ValueError:
            # 忽略重复订阅错误
            pass

    # 4. 注册消息序列化器 (JIT注册)
    for _message_type, serializer in cls._handles_types():
        runtime.add_message_serializer(serializer)

    return agent_type

```python
async def _deliver_message_to_agent(
    self,
    agent_id: AgentId,
    message: Any,
    sender: AgentId | None,
    topic_id: TopicId | None,
    is_rpc: bool,
    message_id: str,
) -> Any:
    """将消息传递给指定代理"""
    
    # 获取或创建代理
    try:
        agent = await self._get_or_create_agent(agent_id)
    except LookupError as e:
        logger.error(f"无法找到代理 {agent_id}: {e}")
        raise UndeliverableException(f"代理不存在: {agent_id}") from e
    
    # 构建消息上下文
    context = MessageContext(
        sender=sender,
        topic_id=topic_id,
        is_rpc=is_rpc,
        cancellation_token=CancellationToken(),
        message_id=message_id,
    )
    
    # 调用代理处理消息
    try:
        result = await agent.on_message(message, context)
        logger.debug(f"代理 {agent_id} 成功处理消息: {message_id}")
        return result
    except CantHandleException as e:
        logger.warning(f"代理 {agent_id} 无法处理消息: {e}")
        raise
    except Exception as e:
        logger.error(f"代理 {agent_id} 处理消息时发生异常: {e}", exc_info=True)
        raise

SingleThreadedAgentRuntime.start() → 
  RunContext.__init__() → 
    asyncio.create_task(self._run()) → 
      while True: _runtime._process_next() → 
        _message_queue.get() → 
          match message_envelope: [SendMessage|PublishMessage|ResponseMessage] → 
            dispatch_to_appropriate_handler()

RunContext.stop() → 
  _stopped.set() → 
    _message_queue.shutdown(immediate=True) → 
      await _run_task → 
        [cleanup] stop_all_background_tasks

RunContext.stop_when_idle() → 
  _message_queue.join() → [等待队列空] → 
    stop() → graceful_shutdown

RunContext.stop_when(condition) → 
  check_condition_periodically() → 
    [condition met] stop() → 
      conditional_shutdown

class RunContext:
    """运行上下文 - 控制运行时的启动和停止"""
    
    def __init__(self, runtime: SingleThreadedAgentRuntime) -> None:
        self._runtime = runtime
        self._run_task = asyncio.create_task(self._run())
        self._stopped = asyncio.Event()
    
    async def _run(self) -> None:
        """主运行循环"""
        while True:
            if self._stopped.is_set():
                return
            await self._runtime._process_next()
    
    async def stop(self) -> None:
        """立即停止运行时"""
        self._stopped.set()
        self._runtime._message_queue.shutdown(immediate=True)
        await self._run_task
    
    async def stop_when_idle(self) -> None:
        """等待队列空闲后停止"""
        await self._runtime._message_queue.join()  # 等待队列为空
        self._stopped.set()
        self._runtime._message_queue.shutdown(immediate=True)
        await self._run_task
    
    async def stop_when(self, condition: Callable[[], bool], check_period: float = 1.0) -> None:
        """满足条件时停止"""
        async def check_condition() -> None:
            while not condition():
                await asyncio.sleep(check_period)
            await self.stop()
        
        await asyncio.create_task(check_condition())

# 运行时使用示例
def start(self) -> RunContext:
    """启动运行时"""
    if self._running:
        raise RuntimeError("运行时已经启动")
    
    self._running = True
    return RunContext(self)

# 智能代理的认知计算模型
class CognitiveArchitecture:
    """认知架构分析"""
    
    def __init__(self):
        # 🧠 Layer 1: 反射层（Reactive Layer）
        self.reactive_layer = {
            'trigger': '@message_handler装饰器',
            'response': '即时类型匹配和条件路由',
            'cognition': '类似大脑的脊髓反射，无需高级思考'
        }
        
        # 🎯 Layer 2: 决策层（Deliberative Layer）  
        self.deliberative_layer = {
            'trigger': 'handler.router()条件判断',
            'response': '基于上下文的智能决策',
            'cognition': '类似大脑的大脑皮层，需要分析和判断'
        }
        
        # 🔮 Layer 3: 元认知层（Meta-cognitive Layer）
        self.meta_cognitive_layer = {
            'trigger': '自定义回复策略和干预处理器',
            'response': '对认知过程本身的认知和控制',
            'cognition': '类似大脑的前额叶皮层，具备自我意识'
        }
    
    def cognitive_flow_analysis(self, message: Any) -> str:
        """认知流程分析"""
        # 认知计算的三阶段流程
        return f"""
        阶段1: 感知输入 → {type(message).__name__} 
        阶段2: 认知处理 → 路由决策和处理器选择
        阶段3: 行为输出 → 生成响应和状态更新
        元认知监控: 整个过程的自我监督和优化
        """

# 消息处理的"意识流"特征
class ConsciousnessStreamProcessing:
    """意识流处理模式"""
    
    def __init__(self):
        # 威廉·詹姆斯的意识流理论在代理系统中的体现：
        # 1. 连续性：消息处理的连续性流动
        # 2. 选择性：路由器的选择性注意
        # 3. 个人性：每个代理的独特处理风格
        # 4. 变化性：动态的处理策略调整
        
        self.consciousness_stream = {
            'continuity': 'asyncio任务的连续处理流',
            'selectivity': 'handler.router()的选择性路由',
            'personality': '每个代理的独特实现',
            'variability': '动态的策略和行为调整'
        }
    
    def stream_of_consciousness_analysis(self, agent_processing_log: List[str]) -> dict:
        """意识流分析"""
        # 通过处理日志分析代理的处理状态
        return {
            'attention_focus': self._analyze_attention_patterns(agent_processing_log),
            'thought_transitions': self._analyze_topic_transitions(agent_processing_log),
            'cognitive_rhythm': self._analyze_processing_rhythm(agent_processing_log)
        }

class CustomReplyStrategy:
    """自定义回复策略 - 基于消息内容智能路由"""
    
    def __init__(self, priority_keywords: List[str], escalation_threshold: int = 3):
        self.priority_keywords = priority_keywords
        self.escalation_threshold = escalation_threshold
        self.message_count = 0
    
    async def __call__(
        self, 
        recipient: Agent, 
        messages: List[BaseChatMessage], 
        sender: Agent, 
        config: Dict[str, Any]
    ) -> Tuple[bool, Optional[str]]:
        """
        自定义回复策略实现
        
        Args:
            recipient: 接收消息的代理
            messages: 消息历史列表
            sender: 发送消息的代理  
            config: 配置参数
            
        Returns:
            Tuple[bool, Optional[str]]: (是否处理, 回复内容)
        """
        
        if not messages:
            return False, None
            
        last_message = messages[-1]
        content = last_message.content.lower()
        
        # 1. 优先级关键词检测
        for keyword in self.priority_keywords:
            if keyword in content:
                priority_response = await self._handle_priority_message(last_message, keyword)
                return True, priority_response
        
        # 2. 情感分析和适应性响应
        sentiment = await self._analyze_sentiment(content)
        if sentiment == "negative" and self.message_count > 2:
            escalation_response = await self._escalate_to_human(last_message)
            return True, escalation_response
        
        # 3. 上下文感知响应
        context = await self._extract_context(messages)
        if context.get("requires_expert"):
            expert_response = await self._route_to_expert(last_message, context)
            return True, expert_response
        
        self.message_count += 1
        return False, None  # 继续默认处理流程
    
    async def _handle_priority_message(self, message: BaseChatMessage, keyword: str) -> str:
        """处理优先级消息"""
        return f"检测到优先级关键词 '{keyword}'，正在优先处理您的请求..."
    
    async def _analyze_sentiment(self, content: str) -> str:
        """情感分析"""
        # 简化的情感分析实现
        negative_indicators = ["生气", "愤怒", "不满", "糟糕", "失望"]
        if any(indicator in content for indicator in negative_indicators):
            return "negative"
        return "neutral"
    
    async def _escalate_to_human(self, message: BaseChatMessage) -> str:
        """升级到人工处理"""
        return "我注意到您可能遇到了一些困难，正在为您转接人工客服..."
    
    async def _extract_context(self, messages: List[BaseChatMessage]) -> Dict[str, Any]:
        """提取对话上下文"""
        context = {
            "topic": None,
            "requires_expert": False,
            "user_intent": None
        }
        
        # 分析最近几条消息确定主题
        recent_content = " ".join([msg.content for msg in messages[-3:]])
        
        if any(tech_word in recent_content for tech_word in ["技术", "代码", "编程", "bug"]):
            context["requires_expert"] = True
            context["topic"] = "technical"
        
        return context

# 使用示例
async def setup_custom_reply_strategy():
    """设置自定义回复策略示例"""
    
    # 创建智能体
    assistant = RoutedAgent("智能助手")
    
    # 创建自定义策略
    custom_strategy = CustomReplyStrategy(
        priority_keywords=["紧急", "重要", "立即", "urgent"],
        escalation_threshold=3
    )
    
    # 注册回复策略
    assistant.register_reply_handler(
        trigger_condition=lambda msg: True,  # 触发条件
        reply_function=custom_strategy,
        priority=0  # 最高优先级
    )
    
    return assistant