AutoGen：会话式多 Agent

AutoGen 的设计哲学

AutoGen（微软研究院开源）的核心思想是：将多 Agent 协作抽象为对话（Conversation）。每个 Agent 都是一个对话参与者，通过消息传递来协作完成任务。

AutoGen 0.4 相比 0.2 进行了全面重构，引入了基于异步消息传递的新架构，支持更灵活的 Agent 拓扑和更好的并发性能。

AutoGen 0.4 架构图

AutoGen 0.4 基础用法

import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models import OpenAIChatCompletionClient

# ── 配置模型客户端 ────────────────────────────────────────
model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    api_key="your-api-key",
    temperature=0
)

# ── 创建 AssistantAgent ───────────────────────────────────
assistant = AssistantAgent(
    name="Python专家",
    model_client=model_client,
    system_message="""你是一位资深 Python 工程师。
    职责：编写高质量、有注释的Python代码，解释设计决策。
    当代码经过测试且运行正常时，在消息末尾写 'TASK_COMPLETE'。
    代码必须包含：类型注解、错误处理、单元测试。"""
)

# ── 创建 UserProxyAgent ───────────────────────────────────
user_proxy = UserProxyAgent(
    name="代码测试员",
    # NEVER：全自动执行，不需要人工输入
    # TERMINATE：执行后询问是否继续
    # ALWAYS：每次都需要人工输入
)

# ── 设置终止条件 ──────────────────────────────────────────
termination = TextMentionTermination("TASK_COMPLETE")

# ── 两人对话 ─────────────────────────────────────────────
async def run_coding_task():
    team = RoundRobinGroupChat(
        participants=[assistant, user_proxy],
        termination_condition=termination,
        max_turns=10
    )

    result = await team.run(
        task="""实现一个线程安全的缓存装饰器 @lru_cache_threadsafe，
        支持 maxsize 参数，并写完整的测试用例。"""
    )

    async for msg in result.messages:
        print(f"[{msg.source}] {msg.content[:200]}\n")

asyncio.run(run_coding_task())

代码执行沙箱

AutoGen 支持多种代码执行环境，生产环境必须使用隔离沙箱：

from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_agentchat.agents import CodeExecutorAgent
import tempfile
from pathlib import Path

# ── 方案1：本地执行（仅开发/测试环境）───────────────────
work_dir = Path(tempfile.mkdtemp())
local_executor = LocalCommandLineCodeExecutor(
    work_dir=work_dir,
    timeout=30,  # 执行超时（秒）
    virtual_env_context=None
)

# ── 方案2：Docker 隔离沙箱（推荐生产环境）──────────────
docker_executor = DockerCommandLineCodeExecutor(
    image="python:3.11-slim",      # 使用最小化镜像
    timeout=60,
    work_dir=work_dir,
    bind_dir=work_dir,             # 挂载工作目录
    auto_remove=True,             # 执行后自动删除容器
)

# ── 创建代码执行 Agent ────────────────────────────────────
async def create_executor_agent():
    async with DockerCommandLineCodeExecutor(
        image="python:3.11-slim",
        timeout=60
    ) as executor:
        code_agent = CodeExecutorAgent(
            name="安全代码执行器",
            code_executor=executor
        )
        yield code_agent

# ── 限制可用的 Python 包 ──────────────────────────────────
# Dockerfile 示例（只安装必要包，减少攻击面）：
# FROM python:3.11-slim
# RUN pip install numpy pandas matplotlib --no-cache-dir
# RUN useradd -m -u 1000 sandbox
# USER sandbox  # 非 root 用户运行

GroupChat：多 Agent 群聊

from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

# 构建专业团队
architect = AssistantAgent(
    name="架构师",
    model_client=model_client,
    system_message="""系统架构专家。负责设计整体方案、
    技术选型、评审代码是否符合架构原则。"""
)

developer = AssistantAgent(
    name="开发者",
    model_client=model_client,
    system_message="""全栈开发工程师。根据架构师的设计编写代码，
    处理具体实现细节，修复 bug。"""
)

tester = AssistantAgent(
    name="测试工程师",
    model_client=model_client,
    system_message="""QA 工程师。编写测试用例，发现边界条件和潜在 bug，
    验证代码质量，当所有测试通过时说 'ALL_TESTS_PASSED'。"""
)

code_executor = CodeExecutorAgent(
    name="代码执行器",
    code_executor=local_executor
)

# SelectorGroupChat：由 LLM 智能决定下一个发言者
termination = (
    TextMentionTermination("ALL_TESTS_PASSED") |
    MaxMessageTermination(30)
)

team = SelectorGroupChat(
    participants=[architect, developer, tester, code_executor],
    model_client=model_client,
    termination_condition=termination,
    selector_prompt="""基于对话历史，选择最合适的下一个发言者。
    当前参与者：{participants}
    对话历史：{history}
    选择规则：设计阶段→架构师；实现阶段→开发者；
              代码生成后→代码执行器；有错误→开发者；验证阶段→测试工程师。
    下一个发言者："""
)

async def run_team():
    result = await team.run(
        task="设计并实现一个基于 Redis 的分布式锁，支持超时和重入。"
    )
    return result

asyncio.run(run_team())

AutoGen vs CrewAI vs LangGraph 三框架对比