🔬 AutoGen 深度源码分析

Microsoft Research | v0.5.x AgentChat + Core | 双层架构：Actor Model Runtime + 高级 AgentChat 抽象

1. 架构总览

AutoGen 采用 双层架构：底层是 autogen-core 基于 Actor Model 的异步消息运行时，上层是 autogen-agentchat 提供高级 Chat Agent 抽象。这种设计让核心运行时可以独立于聊天场景使用（如分布式 Agent 系统），同时 Chat 层为常见对话模式提供便捷 API。

2. Core Runtime — Actor Model

AutoGen Core 实现了一个 Actor Model 风格 的消息运行时。每个 Agent 是一个 Actor，通过 AgentId 唯一标识，通过两种通信模式交互：

2.1 Agent Protocol

# autogen_core/_agent.py — 核心协议
@runtime_checkable
class Agent(Protocol):
    @property
    def metadata(self) -> AgentMetadata: ...
    @property
    def id(self) -> AgentId: ...

    async def on_message(self, message: Any, ctx: MessageContext) -> Any:
        """消息处理器 — Runtime 调用，非 Agent 间直接调用"""
        ...

    async def save_state(self) -> Mapping[str, Any]: ...
    async def load_state(self, state: Mapping[str, Any]) -> None: ...

2.2 SingleThreadedAgentRuntime

核心运行时实现。维护消息队列、订阅关系、Agent 实例池：

3. AgentChat 层 — Chat 语义抽象

AgentChat 在 Core 之上定义了 ChatAgent 协议和 Team 协议，构建完整的对话系统：

# autogen_agentchat/base/_chat_agent.py
class ChatAgent(Protocol):
    @property
    def name(self) -> str: ...       # 唯一标识
    @property
    def description(self) -> str: ... # 供 Team 选择参考

    @property
    def produced_message_types(self) -> Sequence[type[BaseChatMessage]]: ...

    async def on_messages(self, messages, cancellation_token) -> Response: ...
    async def on_reset(self, cancellation_token) -> None: ...

# autogen_agentchat/base/_team.py
class Team(Protocol):
    async def run(self, *, task, cancellation_token) -> TaskResult: ...
    async def run_stream(self, *, task, cancellation_token) -> AsyncGenerator: ...
    async def reset(self) -> None: ...
    async def save_state(self) -> Mapping: ...
    async def load_state(self, state) -> None: ...

3.1 BaseChatAgent — 所有 Chat Agent 的基类

实现了 run() / run_stream() 的默认逻辑，子类只需实现 on_messages() 和 on_messages_stream()：

class BaseChatAgent(ChatAgent, ABC, ComponentBase[BaseModel]):
    async def run(self, *, task, cancellation_token=None) -> TaskResult:
        # 1. 将 task 转换为 input_messages (TextMessage)
        # 2. 调用 on_messages(input_messages, token)
        # 3. 收集 inner_messages + chat_message → TaskResult

    async def run_stream(self, *, task, cancellation_token=None):
        # 1. 同上转换 task
        # 2. 调用 on_messages_stream，逐个 yield 事件
        # 3. 最后 yield TaskResult

4. Agent 体系

4.1 AssistantAgent — 核心 Agent

这是 AutoGen v0.5 的核心 Agent 实现，集成了 LLM 推理、工具调用、Handoff 和结构化输出。

初始化参数一览

on_messages_stream 执行流程

工具调用循环详解

max_tool_iterations 控制顺序工具调用轮次：

# 伪代码：工具调用循环
for loop_iteration in range(max_tool_iterations):
    if isinstance(model_result.content, str):
        return text_response  # 模型返回文本，结束

    # 模型返回 FunctionCall 列表
    results = await asyncio.gather(*[execute_tool(call) for call in calls])

    if has_handoff(results):
        return HandoffMessage  # Handoff 优先

    if loop_iteration < max_tool_iterations - 1:
        model_result = await call_llm()  # 继续循环

4.2 UserProxyAgent — 人类输入代理

通过 input_func 获取人类输入。支持同步/异步函数、CancellationToken 超时取消：

class UserProxyAgent(BaseChatAgent):
    def __init__(self, name, *, input_func=None):
        self.input_func = input_func or cancellable_input  # 默认 stdin

    async def on_messages_stream(self, messages, cancellation_token):
        # 1. 检查是否有 HandoffMessage → 构建提示
        # 2. yield UserInputRequestedEvent
        # 3. await input_func(prompt, cancellation_token)
        # 4. 返回 TextMessage 或 HandoffMessage

4.3 CodeExecutorAgent — 代码执行 Agent

两种模式：被动模式（仅执行收到的代码块）和主动模式（带 model_client 生成+执行+反思）：

class CodeExecutorAgent(BaseChatAgent):
    async def on_messages_stream(self, messages, cancellation_token):
        if model_client is None:
            # 被动模式：从消息中提取代码块执行
            code_blocks = extract_code_blocks(messages)
            result = await execute_code_block(code_blocks)
            return TextMessage(content=result.output)

        # 主动模式：生成 → 执行 → 反思循环
        for nth_try in range(max_retries + 1):
            model_result = await call_llm()
            code_blocks = extract_markdown_code_blocks(model_result)
            if not code_blocks:
                return TextMessage(content=model_result)
            yield CodeGenerationEvent(...)
            result = await execute_code_block(code_blocks)

            if result.exit_code == 0:
                break  # 成功

            # 结构化输出决定是否重试
            retry_decision = await model_client.create(json_output=RetryDecision)
            if not retry_decision.retry:
                break

        # 最终反思
        reflection = await reflect_on_code_results()

5. 消息系统

AutoGen 定义了丰富的消息类型层次，分为 ChatMessage（Agent 间通信）和 AgentEvent（可观测事件）：

关键消息类型

HandoffMessage — Agent 切换的核心

class HandoffMessage(BaseChatMessage):
    target: str              # 目标 Agent 名称
    context: List[LLMMessage] = []  # 附带的 LLM 上下文

    def to_model_message(self) -> UserMessage:
        return UserMessage(content=self.content, source=self.source)

StructuredMessage — 类型安全的输出

class StructuredMessage(BaseChatMessage, Generic[T]):
    content: T  # Pydantic BaseModel 实例

    def to_model_text(self) -> str:
        return self.content.model_dump_json()

6. 工具系统

AutoGen 的工具系统基于 Pydantic 模型自动生成 JSON Schema，支持函数工具和 Workbench 抽象：

6.1 FunctionTool — 自动 Schema 生成

class FunctionTool(BaseTool[BaseModel, BaseModel]):
    def __init__(self, func, description, name=None, strict=False):
        self._signature = get_typed_signature(func)
        # 自动生成 Pydantic 参数模型
        args_model = args_base_model_from_signature(
            func_name + "args", self._signature
        )
        super().__init__(args_model, return_type, name, description, strict)

    async def run(self, args: BaseModel, cancellation_token) -> Any:
        kwargs = {name: getattr(args, name) for name in self._signature.parameters}
        if asyncio.iscoroutinefunction(self._func):
            return await self._func(**kwargs)
        else:
            return await run_in_executor(self._func, **kwargs)

6.2 Workbench — 工具容器抽象

Workbench 是工具的容器接口，提供统一的工具列表和调用 API：

class Workbench(Protocol):
    async def list_tools(self) -> List[ToolSchema]: ...
    async def call_tool(self, name, args, token, call_id) -> ToolResult: ...
    async def call_tool_stream(self, name, args, token, call_id) -> AsyncGenerator: ...

# StaticStreamWorkbench — 包装静态工具列表
# McpWorkbench — 连接 MCP 服务器动态获取工具

7. 群组编排 — Teams

AutoGen 提供 5 种群组编排模式，每种由 BaseGroupChat + 对应的 GroupChatManager 组成：

7.1 BaseGroupChatManager — 群组管理基类

class BaseGroupChatManager(BaseAgent):
    async def _process_messages(self, messages):
        # 1. 验证消息
        await self.validate_group_state(messages)
        # 2. 更新消息线程
        await self.update_message_thread(messages)
        # 3. 检查终止条件
        if await self._check_termination(messages):
            return
        # 4. 选择下一个发言者
        speaker = await self.select_speaker(self._message_thread)
        # 5. 路由消息给选中的 Agent
        await self._route_message(speaker, messages)

7.2 RoundRobinGroupChat — 简单轮转

class RoundRobinGroupChatManager(BaseGroupChatManager):
    async def select_speaker(self, thread):
        current = self._next_speaker_index
        self._next_speaker_index = (current + 1) % len(self._participant_names)
        return self._participant_names[current]

最简单的编排：按参与者顺序轮流。支持嵌套 Team（Team 作为参与者）。

7.3 SelectorGroupChat — LLM 选择发言者

7.4 Swarm — Handoff 驱动

Swarm 编排完全基于 HandoffMessage，不需要中央选择器：

class SwarmGroupChatManager(BaseGroupChatManager):
    async def select_speaker(self, thread):
        # 反向查找最近的 HandoffMessage
        for message in reversed(thread):
            if isinstance(message, HandoffMessage):
                self._current_speaker = message.target
                return [self._current_speaker]
        return self._current_speaker  # 默认继续当前 Agent

7.5 MagenticOne — Ledger 编排

MagenticOne 实现了论文 "Magentic-One: A Generalist Multi-Agent System for Solving Complex Tasks" 的编排策略：

# MagenticOne 进度账本结构 (内部使用 LLM 结构化输出)
class LedgerEntry(BaseModel):
    is_request_completed: bool
    is_in_progress: bool
    is_step_complete: bool
    next_speaker: str | None
    instruction_to_next_speaker: str | None

7.6 DiGraphGroupChat — DAG 图执行 (实验性)

基于有向图 (DAG) 的编排，支持条件边和循环：

class DiGraphEdge(BaseModel):
    target: str
    condition: str | Callable | None  # 条件执行
    activation_group: str = ""       # 前向依赖分组
    activation_condition: Literal["all", "any"] = "all"

class DiGraphNode(BaseModel):
    name: str
    edges: List[DiGraphEdge]  # 出边

# 示例: A → B → C, B 可循环回 A
graph = DiGraph(nodes={
    "A": DiGraphNode(name="A", edges=[DiGraphEdge(target="B")]),
    "B": DiGraphNode(name="B", edges=[
        DiGraphEdge(target="C", condition="exit"),
        DiGraphEdge(target="A", condition="loop"),
    ]),
    "C": DiGraphNode(name="C", edges=[]),
})

8. Handoff 机制 — 深度解析

Handoff 是 AutoGen v0.5 的核心创新之一，它将 Agent 切换建模为工具调用：

class Handoff(BaseModel):
    target: str          # 目标 Agent 名称
    description: str     # 自动生成: "Handoff to {target}."
    name: str            # 自动生成: "transfer_to_{target}"
    message: str         # 自动生成: "Transferred to {target}..."

    @property
    def handoff_tool(self) -> BaseTool:
        def _handoff_tool() -> str:
            return self.message
        return FunctionTool(_handoff_tool, name=self.name,
                           description=self.description, strict=True)

class SocietyOfMindAgent(BaseChatAgent):
    def __init__(self, name, team, model_client, *,
                 instruction=DEFAULT_INSTRUCTION,
                 response_prompt=DEFAULT_RESPONSE_PROMPT):
        self._team = team
        self._model_client = model_client

    async def on_messages_stream(self, messages, cancellation_token):
        # 1. 运行内嵌 Team
        async for msg in self._team.run_stream(task=messages):
            if isinstance(msg, TaskResult):
                result = msg
            else:
                yield msg  # 透传内部事件

        # 2. 用 LLM 生成总结响应
        llm_messages = [
            SystemMessage(self._instruction),  # "Earlier you were asked..."
            *[msg.to_model_message() for msg in inner_messages],
            SystemMessage(self._response_prompt),  # "Output standalone response..."
        ]
        completion = await self._model_client.create(messages=llm_messages)

        # 3. 重置内嵌 Team
        await self._team.reset()

        yield Response(chat_message=TextMessage(content=completion.content))

for nth_try in range(max_retries + 1):
    # 1. LLM 生成代码
    model_result = await model_client.create(messages)
    code_blocks = extract_markdown_code_blocks(model_result)

    if not code_blocks:
        return model_result  # 纯文本，无代码

    # 2. 执行代码 (可被 approval_func 拦截)
    result = await execute_code_block(code_blocks)

    if result.exit_code == 0:
        break  # 成功

    # 3. 错误时：LLM 决定是否重试
    retry_decision = await model_client.create(
        json_output=RetryDecision,  # {retry: bool, reason: str}
    )
    if not retry_decision.retry:
        break

# 4. 最终反思
reflection = await model_client.create(messages + [execution_result])

# 自定义 Context 示例：过滤推理模型的 thought 字段
class ReasoningModelContext(UnboundedChatCompletionContext):
    async def get_messages(self) -> List[LLMMessage]:
        messages = await super().get_messages()
        for message in messages:
            if isinstance(message, AssistantMessage):
                message.thought = None  # 移除冗长的推理过程
        return messages

class Memory(ABC):
    async def update_context(self, model_context) -> UpdateContextResult:
        """根据当前 model_context 查询相关记忆，注入到 context"""
        ...

    async def query(self, query, cancellation_token) -> MemoryQueryResult: ...
    async def add(self, content, cancellation_token) -> None: ...
    async def clear(self) -> None: ...

class ListMemory(Memory):
    def __init__(self, name="ListMemory"):
        self._content: List[MemoryContent] = []

    async def update_context(self, model_context):
        for content in self._content:
            await model_context.add_message(
                SystemMessage(content=content.content)
            )
        return UpdateContextResult(memories=MemoryQueryResult(results=self._content))

    async def add(self, content): self._content.append(content)

# 组合终止条件
termination = (
    HandoffTermination(target="user") |  # Handoff 到人类
    MaxMessageTermination(10) |           # 最多 10 条消息
    TimeoutTermination(60)                # 60 秒超时
)

from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent

model_client = OpenAIChatCompletionClient(model="gpt-4o")
agent = AssistantAgent("assistant", model_client=model_client)
result = await agent.run(task="Hello!")

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import MaxMessageTermination

triage = AssistantAgent("triage", model_client=model_client,
    handoffs=["billing", "support"],
    system_message="Route to billing or support.")
billing = AssistantAgent("billing", model_client=model_client)
support = AssistantAgent("support", model_client=model_client)

team = Swarm([triage, billing, support],
    termination_condition=MaxMessageTermination(5))
await team.run(task="I was charged incorrectly!")

from autogen_agentchat.agents import AssistantAgent, SocietyOfMindAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

writer = AssistantAgent("writer", model_client, system_message="Write well.")
reviewer = AssistantAgent("reviewer", model_client,
    system_message="Review. Say APPROVE when done.")
inner_team = RoundRobinGroupChat([writer, reviewer],
    termination_condition=TextMentionTermination("APPROVE"))

society = SocietyOfMindAgent("writing_team", team=inner_team,
    model_client=model_client)
translator = AssistantAgent("translator", model_client,
    system_message="Translate to Spanish.")

outer_team = RoundRobinGroupChat([society, translator], max_turns=2)
await outer_team.run(task="Write a short story.")