智能体框架实践：AutoGen、AgentScope、CAMEL与LangGraph

原文：Hello-Agents · 第六章 框架实践 AutoGen / LangGraph · 在线阅读

一句话

智能体框架将通用底层逻辑抽象封装，让开发者专注于业务逻辑，实现高效、规范、可靠的智能体应用开发。

什么时候翻这页

• 需要了解主流智能体框架的特点和适用场景
• 希望利用框架提高智能体开发效率
• 构建复杂多智能体协作系统
• 需要解决智能体状态管理、工具调用等底层问题
• 在"涌现式协作"与"显式控制"之间做技术选型

核心概念

智能体框架

提供经过验证的"规范"，抽象和封装智能体共有的重复性工作（主循环、状态管理、工具调用、日志记录等），让开发者专注于独特业务逻辑。

AutoGen

• 核心理念：通过对话实现协作
• 架构特点：分层设计（autogen-core和autogen-agentchat）、异步优先
• 核心组件：
  • AssistantAgent（助理智能体）：任务主要解决者，封装LLM
  • UserProxyAgent（用户代理智能体）：人类用户代言人，执行器
  • RoundRobinGroupChat（轮询群聊）：顺序化对话协调机制

AgentScope

• 核心理念：工程化优先的多智能体平台
• 架构特点：消息驱动架构、组合式设计、分层模块化
• 核心组件：
  • Message（消息）：定义统一消息格式
  • MsgHub（消息中心）：负责智能体间消息路由和状态管理
  • Pipeline：提供灵活的工作流编排能力

CAMEL

• 核心理念：角色扮演（Role-Playing）的协作方法
• 关键技术：引导性提示（Inception Prompting）
• 核心组件：
  • AI User：提出需求、下达指令
  • AI Assistant：执行具体操作和提供解决方案

LangGraph

• 核心理念：将智能体执行流程建模为状态机和有向图
• 核心组件：
  • 节点（Nodes）：代表具体计算步骤
  • 边（Edges）：定义节点间跳转逻辑
  • 全局状态（State）：共享状态对象

怎么做

使用框架开发智能体的基本步骤

1. 明确需求与场景：确定智能体类型、协作方式和复杂度
2. 选择合适框架：根据场景特点选择最适合的框架
3. 设计智能体角色：定义各智能体的职责和能力
4. 实现核心逻辑：
   • AutoGen：定义系统消息、配置群聊规则
   • AgentScope：设计消息传递机制、构建通信网络
   • CAMEL：设计角色提示、定义引导性提示
   • LangGraph：定义状态、创建节点、构建图结构
5. 测试与优化：验证协作流程、调整参数、优化性能

AutoGen软件开发团队实现

1. 配置模型客户端：

from autogen_ext.models.openai import OpenAIChatCompletionClient

def create_openai_model_client():
    return OpenAIChatCompletionClient(
        model=os.getenv("LLM_MODEL_ID", "gpt-4o"),
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1")
    )

2. 定义智能体角色：

def create_product_manager(model_client):
    system_message = """你是一位经验丰富的产品经理..."""
    return AssistantAgent(
        name="ProductManager",
        model_client=model_client,
        system_message=system_message,
    )

3. 定义团队协作流程：

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

team_chat = RoundRobinGroupChat(
    participants=[product_manager, engineer, code_reviewer, user_proxy],
    termination_condition=TextMentionTermination("TERMINATE"),
    max_turns=20,
)

AgentScope三国狼人杀实现

1. 定义消息结构：

from agentscope.message import Msg

message = Msg(
    name="Alice",
    content="Hello, Bob!",
    role="user",
    metadata={
        "timestamp": "2024-01-15T10:30:00Z",
        "message_type": "text",
        "priority": "normal"
    }
)

2. 实现狼人阶段：

async def werewolf_phase(self, round_num: int):
    if not self.werewolves:
        return None
        
    async with MsgHub(
        self.werewolves,
        enable_auto_broadcast=True,
        announcement=await self.moderator.announce(
            f"狼人们，请讨论今晚的击杀目标..."
        ),
    ) as werewolves_hub:
        for _ in range(MAX_DISCUSSION_ROUND):
            for wolf in self.werewolves:
                await wolf(structured_model=DiscussionModelCN)

CAMEL电子书创作实现

1. 初始化角色扮演会话：

from camel.societies import RolePlaying

role_play_session = RolePlaying(
    assistant_role_name="心理学家",
    user_role_name="作家",
    task_prompt=task_prompt,
    model=model,
    with_task_specify=False,
)

2. 运行自动化对话：

chat_turn_limit, n = 30, 0
input_msg = role_play_session.init_chat()

while n < chat_turn_limit:
    n += 1
    assistant_response, user_response = role_play_session.step(input_msg)
    
    print_text_animated(Fore.BLUE + f"作家 (AI User):\n\n{user_response.msg.content}\n")
    print_text_animated(Fore.GREEN + f"心理学家 (AI Assistant):\n\n{assistant_response.msg.content}\n")
    
    if "<CAMEL_TASK_DONE>" in user_response.msg.content or "<CAMEL_TASK_DONE>" in assistant_response.msg.content:
        print(Fore.MAGENTA + "✅ 电子书创作完成！")
        break
    
    input_msg = assistant_response.msg

LangGraph三步问答助手实现

1. 定义全局状态：

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class SearchState(TypedDict):
    messages: Annotated[list, add_messages]
    user_query: str
    search_query: str
    search_results: str
    final_answer: str
    step: str

2. 定义工作流节点：

def understand_query_node(state: SearchState) -> dict:
    user_message = state["messages"][-1].content
    
    understand_prompt = f"""分析用户的查询："{user_message}"
请完成两个任务：
1. 简洁总结用户想要了解什么
2. 生成最适合搜索引擎的关键词

格式：
理解：[用户需求总结]
搜索词：[最佳搜索关键词]"""

    response = llm.invoke([SystemMessage(content=understand_prompt)])
    response_text = response.content
    
    search_query = user_message
    if "搜索词：" in response_text:
        search_query = response_text.split("搜索词：")[1].strip()
    
    return {
        "user_query": response_text,
        "search_query": search_query,
        "step": "understood",
        "messages": [AIMessage(content=f"我将为您搜索：{search_query}")]
    }

3. 构建图：

from langgraph.graph import StateGraph, START, END

def create_search_assistant():
    workflow = StateGraph(SearchState)
    
    workflow.add_node("understand", understand_query_node)
    workflow.add_node("search", tavily_search_node)
    workflow.add_node("answer", generate_answer_node)
    
    workflow.add_edge(START, "understand")
    workflow.add_edge("understand", "search")
    workflow.add_edge("search", "answer")
    workflow.add_edge("answer", END)
    
    memory = InMemorySaver()
    app = workflow.compile(checkpointer=memory)
    return app

命令 / 代码速查

AutoGen核心代码

# 创建模型客户端
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key="your-api-key",
    base_url="https://api.openai.com/v1"
)

# 创建智能体
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent

def create_product_manager(model_client):
    system_message = """你是一位经验丰富的产品经理..."""
    return AssistantAgent(
        name="ProductManager",
        model_client=model_client,
        system_message=system_message,
    )

# 定义团队协作
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

team_chat = RoundRobinGroupChat(
    participants=[product_manager, engineer, code_reviewer, user_proxy],
    termination_condition=TextMentionTermination("TERMINATE"),
    max_turns=20,
)

# 运行团队协作
import asyncio

async def run_software_development_team():
    task = "我们需要开发一个比特币价格显示应用..."
    result = await Console(team_chat.run_stream(task=task))
    return result

if __name__ == "__main__":
    result = asyncio.run(run_software_development_team())

AgentScope核心代码

# 创建消息
from agentscope.message import Msg

message = Msg(
    name="Alice",
    content="Hello, Bob!",
    role="user",
    metadata={"timestamp": "2024-01-15T10:30:00Z"}
)

# 创建智能体
from agentscope.agents import AgentBase

class CustomAgent(AgentBase):
    def __init__(self, name: str, **kwargs):
        super().__init__(name=name, **kwargs)
    
    def reply(self, x: Msg) -> Msg:
        response = self.model(x.content)
        return Msg(name=self.name, content=response, role="assistant")

# 使用消息中心
async with MsgHub(
    participants,
    enable_auto_broadcast=True,
    announcement=announcement_message,
) as hub:
    # 消息交互逻辑

CAMEL核心代码

from camel.societies import RolePlaying
from camel.models import ModelFactory

# 创建模型
model = ModelFactory.create(
    model_platform=ModelPlatformType.QWEN,
    model_type="your-model",
    url="your-api-url",
    api_key="your-api-key"
)

# 初始化角色扮演会话
role_play_session = RolePlaying(
    assistant_role_name="心理学家",
    user_role_name="作家",
    task_prompt=task_prompt,
    model=model,
    with_task_specify=False,
)

# 运行对话
chat_turn_limit, n = 30, 0
input_msg = role_play_session.init_chat()

while n < chat_turn_limit:
    n += 1
    assistant_response, user_response = role_play_session.step(input_msg)
    # 处理响应...
    input_msg = assistant_response.msg

LangGraph核心代码

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 定义状态
class SearchState(TypedDict):
    messages: Annotated[list, add_messages]
    user_query: str
    search_query: str
    search_results: str
    final_answer: str
    step: str

# 定义节点
def understand_query_node(state: SearchState) -> dict:
    # 节点实现...
    return updated_state

# 构建图
def create_search_assistant():
    workflow = StateGraph(SearchState)
    
    workflow.add_node("understand", understand_query_node)
    workflow.add_node("search", search_node)
    workflow.add_node("answer", answer_node)
    
    workflow.add_edge(START, "understand")
    workflow.add_edge("understand", "search")
    workflow.add_edge("search", "answer")
    workflow.add_edge("answer", END)
    
    app = workflow.compile()
    return app

# 运行应用
app = create_search_assistant()
inputs = {"messages": [HumanMessage(content="明天北京天气怎么样？")]}
for event in app.stream(inputs):
    print(event)

与 Python / Claude Code 手册的联系

• Python基础：框架实现依赖Python异步编程、类继承、函数式编程等概念
• Claude Code：框架是对Claude Code中智能体概念的工程化实现，提供更高层次的抽象
• 状态管理：扩展了Python字典和类管理状态的能力，提供更复杂的状态同步机制
• 异步编程：大量使用async/await语法，需要理解Python异步编程模型
• 提示工程：框架使用与Claude Code相同的提示工程原理，但提供了结构化方式

初学者易错点

框架选择误区

• 过度追求功能全面而选择不适合场景的框架
• 忽视框架的工程化需求，导致后期难以维护
• 对"涌现式协作"与"显式控制"的权衡理解不足

AutoGen常见错误

• 系统消息设计过于笼统，导致智能体角色不明确
• 轮询群聊顺序与业务逻辑不匹配
• 忽略终止条件设置，导致对话无限循环

AgentScope常见错误

• 消息结构设计不规范，导致路由错误
• 异步编程概念理解不足，导致阻塞问题
• 分布式部署时未考虑消息顺序性和一致性

CAMEL常见错误

• 引导性提示设计不完善，导致智能体偏离角色
• 任务完成标志使用不当，导致过早或过晚终止
• 双智能体协作扩展到多智能体时缺乏有效协调机制

LangGraph常见错误

• 状态结构设计过于复杂或过于简单
• 条件边逻辑设计不当，导致流程卡死
• 节点间状态传递不一致，导致数据错误

相关词条

• 多智能体协作：多个智能体协同完成复杂任务
• 状态管理：智能体内部状态和交互状态的管理
• 工具调用：智能体使用外部工具扩展能力
• 提示工程：设计有效提示词指导智能体行为
• 异步编程：非阻塞式编程模型，提高并发性能
• 消息驱动架构：基于消息传递的组件交互模式
• 涌现式协作：从简单规则中自然产生的复杂协作行为
• 显式控制：明确定义和控制流程执行路径