介绍

大型语言模型（LLM）的出现重塑了人工智能系统与世界交互和解读世界的方式。传统上，我们采用单代理架构来处理输入、做出决策并产生输出。然而，随着人工智能系统规模的不断扩大，以处理更复杂、多步骤的任务，研究人员和开发人员越来越多地转向多代理系统和基于图的高级架构。这些创新，在LangChain和LangGraph等框架的支持下，使得人工智能系统更加灵活、可扩展且具备协作能力，能够执行复杂的任务。

代理与多代理系统

在人工智能中，代理是一个能够感知环境、做出决策并执行动作以实现目标的自主实体。

多代理系统则由多个在网络中相互作用的自主代理组成，以实现协作目标。这些系统相较于单代理架构具有诸多优势：

• 协作性：多个代理可以汇聚各自的优势，共同处理复杂任务，从而得出更全面的结果。
• 可扩展性：随着工作量的增加，通过添加新代理可以更有效地分配任务，而非仅扩大单个代理的容量。
• 专业化：不同的代理可以专注于特定任务，从而优化性能。例如，一个代理可能负责语言解析，而另一个则负责执行计算或数据检索。

这些特性使得多代理系统成为实现复杂、可扩展且稳健的人工智能解决方案的自然选择。

LangChain与LangGraph

LangChain：构建和管理复杂的LLM链

LangChain是一个开创性的框架，旨在促进LLM（大型语言模型）应用的发展，支持从创建到部署和监控的整个生命周期。LangChain的一个显著优势是支持将多个LLM链接在一起以处理复杂查询。这种链接概念允许一个模型的输出作为另一个模型的输入，从而在LLM之间实现逐步、模块化的决策过程。

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

LangGraph：通过状态化代理增强适应性

在LangChain的基础能力之上，LangGraph在基于图的架构中引入了状态化和自适应的代理。这种创新设计使代理能够从迭代反馈中学习，适应动态环境，并应对日益复杂的挑战。通过利用图中的有向循环，LangGraph允许代理随着时间的推移不断完善其决策过程，有效地模仿了人类通过经验学习的方式。

自适应学习：

LangGraph在其架构中融入了反馈循环，使代理能够回顾和重新评估其先前的行动。通过分析过去的结果，这些代理能够完善策略、减少重复性错误，并动态适应新数据或场景。这种反馈机制增强了系统的长期韧性和有效性。

复杂任务处理：

LangGraph采用“分而治之”的方法，多方面的问题分解为更小、更易管理的组件。专门化的代理负责处理每个组件，并行工作以优化处理时间，并将错误隔离以防止连锁故障。这种模块化方法使系统能够高效地处理错综复杂、多层次的目标。

状态管理：

LangGraph的基础是一个强大的状态管理系统，它能够保存记忆、随动态更新而演进，并允许代理访问共享上下文。这确保了任务之间的连续性，通过减少冗余来优化资源利用，并通过基于上下文的行动来增强决策能力。系统的有状态性支持长期学习，随着时间的推移，积累的专业知识会提高性能。

LangGraph将自适应学习、任务分解和全面的状态管理相结合，为构建智能、可扩展且高效的系统提供了突破性的解决方案，这些系统能够解决现实世界中的挑战。

LangGraph的架构组件

LangGraph框架围绕三个关键组件构建，这些组件能够实现复杂任务的自适应、有状态和模块化工作流程。每个组件在整体架构中都发挥着独特的作用：

节点：任务专门化代理

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

边：定义动态交互

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

状态：共享内存框架

状态作为共享数据结构，捕捉整个工作流程的进展。它充当“全局内存”，保存每个节点计算的结果，并在整个图中提供连续性。这个结构对于节点之间的协作至关重要，因为它允许代理访问并基于先前的结果进行构建。状态使用模式来定义应保留哪些信息，并使用归约器来管理每次计算后的更新。这种设计确保了即使在具有迭代过程的多代理场景中，系统也能保持一致。

通过结合这些组件，LangGraph创建了一个健壮、模块化和自适应的框架，用于编排高级工作流程。它支持动态路由、有状态内存和任务专门化。

用例：

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

1. 客户支持自动化：LangGraph可用于构建高级客户支持系统，通过在专门化代理之间路由请求来处理任务。这些代理可以处理如查询航班信息、更新预订和发放退款等任务。例如，一个代理可以与航班数据库交互，根据客户查询找到可用座位，而另一个代理则可以处理机票更新或取消。
2. 知识检索和问题回答：LangGraph可以通过创建处理、评估和合成文档的代理来自动化信息的检索和评估。例如，一个检索代理搜索相关文档，而一个评估代理则判断检索到的信息是否有效地回答了用户的问题。这可以应用于研究助理或基于内容的决策制定过程中。
3. 构建具有记忆的自主代理：LangGraph允许创建能够在交互中保留记忆的自我改进代理。在自主系统中，代理可以使用内存模块来存储先前的经验，使它们能够合成信息并随时间调整其行为。这样的用例在个人助理应用中非常有价值，代理可以在其中学习和适应用户的偏好。
4. 业务流程自动化：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, MessagesPlaceholder
from langgraph.graph import END, StateGraph

第二步：创建代理

代理是一个能够使用工具执行任务的助手。这里有一个可复用的函数来创建代理：

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)

第三步：定义工具

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

from typing import Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_experimental.utilities import PythonREPL

# Search tool
tavily_tool = TavilySearchResults(max_results=3)
# Local Python code execution tool
repl = PythonREPL()
@tool
def 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)}"

第四步：创建并连接代理

我们定义了两个代理：一个用于研究，另一个用于图表生成。每个代理都配备了其相应的工具。

from langchain_openai import ChatOpenAI
import functools

llm = ChatOpenAI(model="gpt-3.5-turbo")
# Research agent
research_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 agent
chart_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")

第五步：构建工作流

使用StateGraph，我们定义了一个动态工作流，使代理能够进行协作。

workflow = StateGraph(AgentState)

# Add nodes for agents
workflow.add_node("Researcher", research_node)
workflow.add_node("chart_generator", chart_node)
workflow.add_node("call_tool", tool_node)
# Add conditional edges between nodes
workflow.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 point
workflow.set_entry_point("Researcher")
graph = workflow.compile()

第六步：调用系统

最后，我们通过提供一个人类查询来调用系统。

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 outputs
events = graph.stream(
    {"messages": [human_message]},
    {"recursion_limit": 20},
)
for event in events:
    # Print messages from agents
    for message in event.get("messages", []):
        print(message)

交互示例

现在，让我们来看一个基于用户请求的示例：获取2015年至2020年美国国内生产总值（GDP）的数据，并创建一个图表来可视化这些数据。这个示例展示了多代理系统如何无缝地自动化数据检索和可视化。

研究代理在利用Tavily工具进行在线数据检索后，处理获得的信息，并将其转发给图表生成代理。这确保了代理之间的无缝协作，以高效地满足用户的需求。

图表生成代理接收到数据后，会编写生成图表所需的代码。通过动态创建可视化图表，它确保了用户的请求能够高效且准确地完成。