对比5个最需要了解的AI多智能体编排框架【2】：LangGraph篇 - 文章 - 开发者社区

PART 1

picture.image

点击上方蓝字关注我们

上篇（对比5个最需要了解的AI多智能体编排框架，你将如何选择？【1】）中我们介绍了多智能体系统及5个常用编排框架，并使用OpenAI最近推出的轻量级框架Swam实现了一个简单的多智能体系统，由一个负责监管的Agent与两个任务执行的Agent组成：

picture.image

简单回顾Swam这个小框架的特点：

简单，十分钟学会
仅支持OpenAI模型
复杂的智能体工作流难以胜任
灵活性差，基本没有可扩展性

显然， 当前的Swam还只适合用来创建多Agent的演示与原型应用，但暂时不要考虑在生产环境中使用。

本篇让我们继续研究相同的多智能体系统在LangGraph这个框架中的实现，以认识不同框架的特点与适用场景。

快速认识LangGraph

LangGraph是著名LLM底层开发框架LangChain公司推出的独立框架，主要 面向复杂的RAG、智能体与多智能体应用 开发。用过LangChain或者LlamaIndex开发的朋友可能都知道，在这两个主流框架中，早期其实都内置有开箱即用的更简单的Agent开发组件。比如在先前LangChain中，创建一个会使用工具（Tools）的ReAct范式的Agent并不复杂：把定义好的tools交给大模型，然后让大模型自行规划与选择工具的使用，以完成输入任务。如下：


        
            

          ...  
#准备tools，model，prompt  
search = TavilySearchResults(max\_results=2)  
@tool  
def email(topic: str) -> str:  
  return 'Email completed.'  
  
tools = [search, email]  
model = ChatOpenAI(model="gpt-4o")  
prompt = hub.pull("hwchase17/openai-functions-agent")  
  
#创建agent  
agent = create\_tool\_calling\_agent(model, tools, prompt)  
  
#调用agent  
agent\_executor = AgentExecutor(agent=agent, tools=tools)  
agent\_executor.invoke({"input": "搜索明天的天气发送到mm@aa.com"})

那为什么还需要LangGraph？两个核心的原因是：

为了支持更复杂的LLM应用，特别是需要循环迭代的工作流，以及需要多智能体协作与交互的应用。 比如Self-RAG ：

picture.image

为了让AI智能体更加可控与可预测。 尽管LLM已经足够强大，但完全依赖于其自行规划与决策行动的“黑盒”智能体仍然存在较大的不确定性，这会极大的限制其在生产环境，特别是企业环境中的应用。

因此，LangGraph推出的意义（包括LlamaIndex最新的Workflows）就在于 通过对LLM应用流程与状态的细粒度控制，以创建更加可靠、可控、可预测的RAG或者智能体应用。 所以LangGraph是一个相对底层的框架，功能强大，但也较复杂。

用LangGraph实现多智能体

LangGraph的核心方法是：通过定义一个Graph图结构的流程来代表你需要创建的LLM应用工作流。Graph的特点决定了你的应用具备了极大的灵活性：

支持并行、条件分支、循环等各种细粒度工作流控制
灵活定义每个节点任务，简单函数、LLM调用，或一次Agent交互
可持久的全局状态控制，工作流可随时暂停、启动或介入人工交互

因此，基于LangGraph创建Agent应用的主要工作就是定义这个Graph中的节点（node，代表工作环节）与边（edge，代表不同环节之间的关系）。

我们用LangGraph来实现前面简单的多智能体系统：

基础模型与State定义

定义使用的大模型，这里统一使用gpt-4o-mini。当然，多智能体系统的一个好处是：你可以在不同的Agent中使用不同的模型。


        
            

          #llm  
llm = ChatOpenAI(model="gpt-4o-mini")  
  
#State（LangGraph上下文结构）  
class AgentState(TypedDict):  
    messages: Annotated[Sequence[BaseMessage], operator.add]  
    next: str

2. 定义工具（两个tools）

这是最基础的Agent概念，我们用之前类似的方式实现两个tool：一个用于网络搜索，一个模拟邮件发送。


        
            

          #两个Agent的工具  
tavily\_tool = TavilySearchResults(max\_results=5)  
  
@tool  
def mail\_tool(subject: str, body: str,recipient: str = None) -> str:  
    """用于发送邮件"""  
  
    recipient = recipient or "yancanping@cmclouds.com"  
    print(f"开始发送邮件，收件人：{recipient}，主题：{subject}，内容：{body}")  
    return f"Sent email to {recipient} with subject '{subject}' and body '{body}'"

3. 定义Agent及其对应节点

在这个需要实现的分层多智能体系统中，存在三个Agent。一个是管理Agent，另外两个是工作Agent。（这里仅为了演示多智能体系统，实际中这里的两个简单工作节点也可以是直接的工具调用而非Agent调用）


        
            

          # 定义三个可用的AI Agent  
def supervisor\_agent(state):  
    messages\_content = "\n".join([f"{msg.name}: {msg.content}" for msg in state['messages']])  
    prompt = (  
        "评估用户问题并拆分任务,判断下一个需要的AI助手。AI助手包括Researcher、Emailer。如果无需AI助手，请回复：FINISH。\n"  
        "消息内容：\n"  
        f"{messages\_content}\n"  
        "下一步应该由谁来行动？从以下选项中选择：['FINISH', 'Researcher', 'Emailer']，注意不要有其它字符。"  
    )  
      
    response = llm.invoke(prompt)  
    next\_action = response.content.strip()   
      
    return {"next": next\_action}  
  
def research\_agent(state):  
    research\_agent = create\_react\_agent(llm, tools=[tavily\_tool])  
    result = research\_agent.invoke(state)  
    return {  
        "messages": [HumanMessage(content=result["messages"][-1].content, name="Researcher")]  
    }  
  
def email\_agent(state):  
    email\_agent = create\_react\_agent(llm, tools=[mail\_tool])  
    result = email\_agent.invoke(state)  
    return {  
        "messages": [HumanMessage(content=result["messages"][-1].content, name="Emailer")]  
    }

4. 定义Graph（工作流）

Agent准备完成后，可以创建完整的工作流（Graph），添加其中的node和edge即可。


        
            

          # 定义工作流  
workflow = StateGraph(AgentState)  
workflow.add\_node("Supervisor", supervisor\_agent)  
workflow.add\_node("Researcher", research\_agent)  
workflow.add\_node("Emailer",email\_agent)  
  
workflow.add\_edge(START, "Supervisor")  
workflow.add\_edge('Researcher','Supervisor')  
workflow.add\_edge('Emailer','Supervisor')  
workflow.add\_conditional\_edges("Supervisor", lambda x: x["next"], {'Researcher': 'Researcher', 'Emailer': 'Emailer', 'FINISH': END})  
  
graph = workflow.compile()

6. 测试应用

现在来测试这个简单的多智能体系统：


        
            

          for s in graph.stream(  
    {  
        "messages": [  
            HumanMessage(content="搜索明天的南京天气情况,发送邮件给yancp@aa.com",name="User"),  
        ]  
    }  
):  
    if "\_\_end\_\_" not in s:  
        print(s)  
        print("--------------------")

在下图的输出结果中你可以很清晰的看到完整的Agent之间的协作过程：Supervisor首先判断需要调用Researcher这个Agent，在获得结果后，判断下一步需要使用Emailer这个Agent，最后判断可以结束任务（FINISH）：

picture.image