AutoGen：对话驱动的多Agent框架

微软研究院的AutoGen开创了"对话即协作"的Agent范式。与LangGraph的图驱动不同，AutoGen将多Agent协作建模为一组Agent之间的对话，每个Agent有独立的角色和能力。

核心架构

Agent类型体系

from autogen import (
    ConversableAgent, AssistantAgent, UserProxyAgent,
    GroupChat, GroupChatManager
)

# AssistantAgent: AI助手，有系统消息定义角色
researcher = AssistantAgent(
    name="Researcher",
    system_message="""你是一位AI研究分析师。
    职责：
    1. 分析用户需求
    2. 搜索和整理相关信息
    3. 提供结构化的分析报告
    
    约束：
    - 基于事实，不编造
    - 注明信息来源
    - 区分事实和推测
    """,
    llm_config={"model": "gpt-4o", "temperature": 0.3},
    tools=[web_search, knowledge_base_search]
)

# UserProxyAgent: 用户代理，可以执行代码
user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",  # 不等待人类输入
    code_execution_config={
        "work_dir": "workspace",
        "use_docker": True,  # 安全执行环境
        "timeout": 60
    }
)

对话管理

class ConversationManager:
    def __init__(self):
        self.conversations = {}
        self.termination_conditions = []
    
    def add_termination(self, condition):
        """添加对话终止条件"""
        self.termination_conditions.append(condition)
    
    def check_termination(self, messages):
        for condition in self.termination_conditions:
            if condition.check(messages):
                return True
        return False

# 常见终止条件
class MaxRoundsTermination:
    def __init__(self, max_rounds=10):
        self.max_rounds = max_rounds
    
    def check(self, messages):
        return len(messages) >= self.max_rounds

class KeywordTermination:
    def __init__(self, keywords):
        self.keywords = keywords
    
    def check(self, messages):
        if messages:
            return any(kw in messages[-1]["content"] for kw in self.keywords)
        return False

多Agent协作模式

模式1：顺序对话

def sequential_conversation(task):
    """Agent按顺序处理任务"""
    # Agent 1: 分析需求
    analysis = analyst.generate(f"分析以下任务：{task}")
    
    # Agent 2: 编写代码
    code = coder.generate(f"基于以下分析编写代码：{analysis}")
    
    # Agent 3: 审查代码
    review = reviewer.generate(f"审查以下代码：{code}")
    
    # Agent 4: 优化代码
    if "问题" in review:
        final_code = coder.generate(f"根据审查意见优化代码：{review}")
    else:
        final_code = code
    
    return final_code

模式2：群聊协作

def group_chat_collaboration(task):
    """多Agent群聊协作"""
    agents = [
        UserProxyAgent(name="User", human_input_mode="NEVER"),
        AssistantAgent(name="Planner", system_message="负责制定计划"),
        AssistantAgent(name="Coder", system_message="负责编写代码"),
        AssistantAgent(name="Tester", system_message="负责测试"),
        AssistantAgent(name="Reviewer", system_message="负责审查")
    ]
    
    group_chat = GroupChat(
        agents=agents,
        messages=[],
        max_round=20,
        speaker_selection_method="auto"  # 自动选择下一个发言者
    )
    
    manager = GroupChatManager(
        groupchat=group_chat,
        llm_config={"model": "gpt-4o"}
    )
    
    agents[0].initiate_chat(manager, message=task)

模式3：嵌套对话

def nested_conversation(task):
    """Agent内部发起子对话"""
    
    # 主Agent处理任务
    main_agent = AssistantAgent(
        name="Main",
        system_message="你是项目经理，可以委托子任务给其他Agent"
    )
    
    # 当主Agent遇到需要深入研究的子问题时
    # 它可以发起一个子对话
    def research_subtask(subtask):
        researcher = AssistantAgent(
            name="Researcher",
            system_message="你是研究员，擅长信息检索"
        )
        result = researcher.generate(f"研究：{subtask}")
        return result
    
    # 主Agent可以在处理过程中调用子对话
    main_agent.register_function(
        function_map={"research": research_subtask}
    )

代码执行环境

安全执行

class SafeCodeExecutor:
    def __init__(self):
        self.docker_config = {
            "image": "python:3.11-slim",
            "timeout": 60,
            "memory_limit": "512m",
            "cpu_limit": 1.0,
            "network": "none",  # 禁止网络访问
        }
        self.allowed_packages = [
            "numpy", "pandas", "matplotlib", 
            "scipy", "scikit-learn"
        ]
    
    def execute(self, code):
        # 1. 静态检查
        issues = self._static_check(code)
        if issues:
            return {"error": "代码检查未通过", "issues": issues}
        
        # 2. Docker执行
        result = self._docker_exec(code)
        
        return result
    
    def _static_check(self, code):
        """静态安全检查"""
        forbidden = [
            "import os", "import subprocess", "import socket",
            "open(", "__import__", "eval(", "exec("
        ]
        issues = []
        for pattern in forbidden:
            if pattern in code:
                issues.append(f"禁止使用: {pattern}")
        return issues

代码执行反馈循环

def code_feedback_loop(agent, task, max_attempts=3):
    """代码编写-执行-修正的反馈循环"""
    for attempt in range(max_attempts):
        # Agent生成代码
        code = agent.generate(f"任务：{task}\n尝试：{attempt+1}")
        
        # 执行代码
        result = executor.execute(code)
        
        if result["success"]:
            return code, result["output"]
        
        # 反馈错误，让Agent修正
        feedback = f"""
        代码执行失败：
        错误信息：{result['error']}
        请修正代码。
        """
        task = task + "\n\n" + feedback
    
    return None, "达到最大尝试次数"

高级特性

Agent可序列化

def save_agent_state(agent, path):
    """保存Agent状态，支持恢复"""
    state = {
        "name": agent.name,
        "system_message": agent.system_message,
        "llm_config": agent.llm_config,
        "chat_history": agent.chat_messages,
        "registered_tools": list(agent.tools.keys())
    }
    with open(path, 'w') as f:
        json.dump(state, f)

def load_agent_state(path):
    """从文件恢复Agent"""
    with open(path) as f:
        state = json.load(f)
    agent = AssistantAgent(
        name=state["name"],
        system_message=state["system_message"],
        llm_config=state["llm_config"]
    )
    agent.chat_messages = state["chat_history"]
    return agent

自定义Agent行为

class CustomAgent(ConversableAgent):
    def __init__(self, name, **kwargs):
        super().__init__(name, **kwargs)
        self.register_hook("process_message_before_send", self._preprocess)
        self.register_hook("process_message_after_receive", self._postprocess)
    
    def _preprocess(self, message):
        """发送前预处理"""
        # 添加时间戳
        message["content"] = f"[{datetime.now()}] {message['content']}"
        return message
    
    def _postprocess(self, message):
        """接收后处理"""
        # 记录消息日志
        self._log(message)
        return message
    
    def _log(self, message):
        """消息日志"""
        with open("agent_log.jsonl", 'a') as f:
            f.write(json.dumps({
                "timestamp": datetime.now().isoformat(),
                "sender": message.get("from"),
                "content": message["content"][:200]
            }) + "\n")

实际案例：数据分析Agent

def build_data_analysis_agent():
    """构建数据分析Agent系统"""
    
    # 数据科学家Agent
    data_scientist = AssistantAgent(
        name="DataScientist",
        system_message="""你是数据科学家，负责：
        1. 理解用户的数据分析需求
        2. 编写Python代码进行数据分析
        3. 解释分析结果
        
        使用pandas, numpy, matplotlib进行数据分析。
        确保代码包含异常处理和数据验证。
        """,
        llm_config={"model": "gpt-4o"}
    )
    
    # 代码执行Agent
    code_runner = UserProxyAgent(
        name="CodeRunner",
        human_input_mode="NEVER",
        code_execution_config={
            "work_dir": "analysis_workspace",
            "use_docker": "python:3.11-slim"
        },
        system_message="你负责执行代码并返回结果。不生成代码，只执行。"
    )
    
    # 启动分析
    code_runner.initiate_chat(
        data_scientist,
        message="""分析销售数据：
        1. 读取 /data/sales.csv
        2. 按月统计销售趋势
        3. 找出Top 10产品
        4. 生成可视化图表
        5. 输出分析报告
        """
    )

AutoGen vs 其他框架

维度AutoGenLangGraphCrewAI
核心范式对话驱动图驱动角色驱动
代码执行内置Docker需自定义需自定义
多Agent原生支持需手动编排支持
状态管理对话历史检查点任务上下文
适合场景研究探索生产系统快速原型

结语

AutoGen将多Agent协作还原为最自然的形式——对话。它的优势在于代码执行能力和灵活的对话管理。劣势在于控制精度不如图驱动框架。对于需要多专家协作探索的研究型任务，AutoGen是最佳选择。对于需要精确控制执行流程的生产系统，LangGraph更合适。选择框架的关键是匹配你的任务特性。