PART 1

点击上方蓝字关注我们

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

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

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

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

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

1

快速认识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 ：

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

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

2

用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("--------------------")