带LangGraph的多智能体工作流

398 阅读 0 留言 0 点赞

介绍
大型语言模型（LLMs）的出现重塑了AI系统与世界互动和解释的方式。传统上，单个智能体架构被用来处理输入、做出决策并产生输出。然而，随着AI系统规模的扩大，以处理更多复杂、多步骤的任务，研究人员和开发人员越来越多地转向多智能体系统和先进的图结构架构。得益于LangChain和LangGraph等框架的支持，这些创新使得更加灵活、可扩展且协作的AI系统能够执行复杂的任务。

智能体与多智能体系统
AI中的智能体是一个能够感知其环境、做出决策并执行行动以达成目标的自主实体。

多智能体系统包括多个在联网环境中互动的自主智能体，以实现协作目标。相比于单个智能体架构，这些系统提供了若干优势：

协作：多个智能体可以汇集各自的优势，共同完成复杂的任务，从而产生更全面的结果。
可扩展性：随着工作负载的增长，添加新的智能体可以比扩展单个智能体的能力更有效地分配任务。
专业化：不同的智能体可以专攻特定任务，优化性能。例如，一个智能体可以处理语言解析，而另一个则执行计算或数据检索。
这些特性使得多智能体系统成为实现复杂、可扩展且稳健的AI解决方案的自然选择。

LangChain & LangGraph

LangChain：构建和管理复杂的LLM链
LangChain是一个开创性的框架，旨在促进LLM应用程序的开发，支持从创建到部署和监控的整个生命周期。LangChain的一个显著优势是它支持将多个LLM链接在一起处理复杂查询。这一链接概念允许一个模型的输出作为另一个模型的输入，从而实现在LLM之间进行阶梯式、模块化的决策过程。

LangChain使用有向无环图（DAGs）来构建这些链。在DAG架构中，LLMs在一个有向图中运行，每个节点代表一个特定的处理或决策过程，且没有可能导致无限循环的循环。这种架构特别适合处理日益复杂和多步骤的任务，因为它提供了从输入到输出的清晰线性路径。

LangGraph：增强适应能力的有状态智能体
基于LangChain的基础能力，LangGraph在基于图的架构中引入了有状态和适应性智能体。这种创新设计使智能体能够从迭代反馈中学习，适应动态环境，并应对日益复杂的挑战。通过利用图中的有向循环，LangGraph允许智能体随着时间的推移逐步改进其决策过程，有效地模拟人类通过经验学习的过程。

自适应学习：
LangGraph在其架构中引入了反馈循环，使智能体能够回顾并重新评估其先前的行为。通过分析过去的成果，这些智能体会改进策略，减少重复错误，并根据新数据或场景动态调整。这种反馈机制增强了系统的长期弹性和有效性。
处理复杂任务：
LangGraph采用“分而治之”的方法，将大型、多方面的复杂问题分解为较小、可管理的组件。专门化的智能体处理每个组件，并行工作以优化处理时间，并将错误隔离以防止连锁反应。这种模块化的方法使系统能够高效地处理复杂的多层目标。
状态管理：
强大的状态管理系统构成了LangGraph的基础，保留内存，随着动态更新而演变，并使智能体能够访问共享上下文。这确保了在任务之间的一致性，通过减少冗余来优化资源利用，并通过上下文相关的行为来增强决策。系统的有状态性支持长期学习，累积的经验随着时间的推移提高性能。
LangGraph将自适应学习、任务分解和全面的状态管理整合在一起，成为构建智能、可扩展且高效的系统以解决实际挑战的开创性解决方案。

LangGraph架构组件
LangGraph框架围绕三个关键组件构建，这些组件使复杂任务的自适应、有状态和模块化工作流程的创建成为可能。每个组件在整体架构中扮演着独特的作用：

节点：任务专门化的智能体
节点是LangGraph的基本构建块。每个节点代表一个自主智能体，设计用于执行特定功能，如处理数据、与API交互或进行决策。这些智能体独立运行，这意味着每个节点都专门用于其任务，并可在不同的配置中重复使用。例如，一个节点可能负责清理和标准化输入数据，而另一个节点可能生成图表或图形等视觉输出。这种模块化确保了复杂工作流程的灵活性和可扩展性。

边：定义动态交互
边形成了节点之间的连接，决定了数据和任务如何在图中流动。这些连接可以是静态的或条件性的，使系统能够根据其状态进行动态调整。例如，一个条件边可能会将进程导向工具节点，如果调用了特定工具或提供了最终答案则结束工作流。这种适应性使LangGraph能够响应上下文变化，优化资源使用，并确保任务的逻辑进展。

状态：共享内存框架
状态作为共享数据结构，捕获整个工作流程的进度。它充当“全局内存”，保留每个节点计算的结果，并在图中提供连续性。这种结构对于节点之间的协作至关重要，因为它允许智能体访问和构建先前的结果。状态使用模式来确定应保留的信息，并使用更新程序在每次计算后管理更新。这种设计确保了即使在涉及多智能体场景和迭代过程的情况下，系统也保持一致。

通过组合这些组件，LangGraph创建了一个稳健、模块化且适应性强的高级工作流程协调框架。它支持动态路由、有状态内存和任务特定的专门化。

应用场景
LangGraph是一个强大的框架，通过创建相互连接的智能体（或节点）图来实现复杂自动化和AI工作流程。以下是LangGraph可以有效应用的一些应用场景：

客户支持自动化：LangGraph可用于构建高级客户支持系统，通过将请求路由到专门的智能体之间来实现。这些智能体可以处理查询航班信息、更新预订和提供退款等任务。例如，一个智能体可以与航班数据库交互，基于客户的查询找到可用座位，而另一个智能体可以处理票务更新或取消。
知识检索和问答：LangGraph可以通过创建处理、分级和综合文档的智能体来自动化信息检索和评估。例如，检索智能体搜索相关文档，而分级智能体评估检索到的信息是否有效回答了用户的问题。这可以应用于如研究助手或基于内容的决策过程系统中。
构建具有记忆的自主智能体：LangGraph允许创建在交互中保留记忆的自我改进智能体。在自主系统中，智能体可以使用记忆模块来存储先前的经验，允许它们综合信息并随着时间的推移调整其行为。这种用例在个人助理应用中很有价值，智能体可以学习并适应用户的偏好。
业务流程自动化：LangGraph可以通过连接专注于不同任务的各种智能体来自动化多步骤业务流程。这些任务可以从处理用户请求到与外部API交互以及更新中央数据库。一个实用的例子是使用LangGraph在企业内部自动化工作流程，如管理客户订单、库存系统或资源分配。
其中一些应用场景由多智能体系统的支持实现。让我们简要地考察三种最重要的类型。

多智能体系统架构
实现多智能体系统（MAS）可以采取多种形式，这取决于任务的复杂性和要求：

协作架构：在这种架构中，两个或多个智能体协作以实现单一目标。它们可以访问共享工具，如API或数据库，以检索相关信息并生成协调响应。这适用于对单个智能体而言过于复杂但无需广泛协调的任务。
监督架构：在更为结构化的安排中，监督智能体负责监督和协调从属智能体的行为，确保每个组件任务正确执行。监督智能体评估输出，并可能根据从属智能体的进度调整工作流。这种方法在需要高可靠性和复杂决策的场景中很有价值。
分层架构：这种高级架构引入了多层智能体，每层专注于特定领域。在层级顶端，监督团队协调智能体组。这种方法可扩展、灵活，并反映了AI向更多自主决策系统发展的趋势。

LangGraph Studio
智能体框架的一个常见挑战是理解在智能体迭代过程中发生了什么。这就是 LangGraph Studio 发挥作用的地方。它提供了一个专门的智能体集成开发环境 (IDE)，允许开发人员可视化、交互并调试复杂的智能体工作流。

借助其视觉图形和状态编辑功能，LangGraph Studio 帮助用户更好地理解智能体流程并加速迭代。它还与 LangSmith 集成，使团队成员能够协作调试故障模式。

以下是一个 LangGraph Studio 实际应用的示例。

实战
在这个实战教程中，我们将创建一个多智能体系统，该系统能够在线抓取数据并基于这些数据生成图表。我们将逐步定义智能体、工具以及将所有内容整合到一个功能工作流中的逻辑。让我们开始实现吧！

第一步：导入所需库
我们首先导入构建智能体和管理工作流所需的模块。

from langchain_core.messages import (    BaseMessage,    HumanMessage,    ToolMessage,)from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langgraph.graph import END, StateGraph

步骤2：创建智能体
智能体是一种能够使用工具来执行任务的助手。这里有一个可重用的函数来创建智能体：

def create_agent(llm, tools, system_message: str):    """Create an agent."""    tool_names = ", ".join(tool.name for tool in tools)    system_msg = (        "You are a helpful AI assistant, collaborating with other assistants. Use the provided tools "        "to progress towards answering the question. If you are unable to fully answer, that's OK, another "        "assistant with different tools will help where you left off. Execute what you can to make progress. "        "If you or any of the other assistants have the final answer or deliverable, prefix your response with "        "FINAL ANSWER so the team knows to stop. You have access to the following tools, if you provide a tool "        "call, you must choose one of the following tools, you can't provide a name that isn't in this list: "        "{tool_names}.\n{system_message}"    )    prompt = ChatPromptTemplate.from_messages(        [            ("system", system_msg),            MessagesPlaceholder(variable_name="messages"),        ]    ).partial(system_message=system_message, tool_names=tool_names)    return prompt | llm.bind_tools(tools)

步骤3：定义工具
在这里，我们定义两个工具：一个用于在线搜索，另一个用于执行Python代码以生成图表。

from typing import Annotatedfrom langchain_community.tools.tavily_search import TavilySearchResultsfrom langchain_experimental.utilities import PythonREPL

# Search tooltavily_tool = TavilySearchResults(max_results=3)# Local Python code execution toolrepl = PythonREPL()@tooldef python_repl(code: Annotated[str, "The python code to execute to generate your chart."]):    """Execute python code. Print output to see the result, which will be visible to the user."""    try:        result = repl.run(code)        return f"Successfully executed:\n```python\n{code}\n```\nStdout: {result}\n\nIf you have completed all tasks, respond with FINAL ANSWER."    except BaseException as e:        return f"Failed to execute. Error: {repr(e)}"

步骤4：创建并连接智能体
我们定义了两个智能体：一个用于研究，另一个用于图表生成。每个智能体都配备了相应的工具。

from langchain_openai import ChatOpenAIimport functools

llm = ChatOpenAI(model="gpt-3.5-turbo")# Research agentresearch_agent = create_agent(    llm,    [tavily_tool],    system_message="You should provide accurate data. You must invoke a tool name that is in the list of tools available; you can't use chart_generator as a tool name.",)research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")# Chart generator agentchart_agent = create_agent(    llm,    [python_repl],    system_message="Any charts you display will be visible by the user.",)chart_node = functools.partial(agent_node, agent=chart_agent, name="chart_generator")

步骤5：构建工作流
使用StateGraph，我们定义了一个动态工作流，允许智能体进行协作。

workflow = StateGraph(AgentState)

# Add nodes for agentsworkflow.add_node("Researcher", research_node)workflow.add_node("chart_generator", chart_node)workflow.add_node("call_tool", tool_node)# Add conditional edges between nodesworkflow.add_conditional_edges(    "Researcher",    router,    {"continue": "chart_generator", "call_tool": "call_tool", "__end__": END},)workflow.add_conditional_edges(    "chart_generator",    router,    {"continue": "Researcher", "call_tool": "call_tool", "__end__": END},)workflow.add_conditional_edges(    "call_tool",    lambda x: x["sender"],    {"Researcher": "Researcher", "chart_generator": "chart_generator"},)# Set entry pointworkflow.set_entry_point("Researcher")graph = workflow.compile()

步骤6：调用系统
最后，我们通过提供人类的查询来调用系统。

from langchain_core.messages import HumanMessage

human_message = HumanMessage(    content="Fetch the USA's GDP from 2015 to 2020, obtain data year by year, then draw a line graph of it. Once you code it up, finish.")# Stream events and visualize outputsevents = graph.stream(    {"messages": [human_message]},    {"recursion_limit": 20},)for event in events:    # Print messages from agents    for message in event.get("messages", []):        print(message)