当前位置： 首页 > article >正文

【LangGraph Agent架构篇—多智能体系统1】【多智能体网络】

article 2025/3/4 9:13:13

前言

多智能体网络是处理复杂任务的一种有效方法是，分而治之，即为每个任务创建一个Agent。

一、LangGraph

1-1、介绍

在这里插入图片描述

LangGraph是一个专注于构建有状态、多角色应用程序的库，它利用大型语言模型（LLMs）来创建智能体和多智能体工作流。这个框架的核心优势体现在以下几个方面：

  • 周期性支持：LangGraph允许开发者定义包含循环的流程，这对于大多数中智能体架构来说至关重要。这种能力使得LangGraph与基于有向无环图（DAG）的解决方案区分开来，因为它能够处理需要重复步骤或反馈循环的复杂任务。
  • 高度可控性：LangGraph提供了对应用程序流程和状态的精细控制。这种精细控制对于创建行为可靠、符合预期的智能体至关重要，特别是在处理复杂或敏感的应用场景时。
  • 持久性功能：LangGraph内置了持久性功能，这意味着智能体能够跨交互保持上下文和记忆。这对于实现长期任务的一致性和连续性非常关键。持久性还支持高级的人机交互，允许人类输入无缝集成到工作流程中，并使智能体能够通过记忆功能学习和适应。

1-2、特点

1. Cycles and Branching（循环和分支）

  • 功能描述：允许在应用程序中实现循环和条件语句。
  • 应用场景：适用于需要重复执行任务或根据不同条件执行不同操作的场景，如自动化决策流程、复杂业务逻辑处理等。

3. Persistence（持久性）

  • 功能描述：自动在每个步骤后保存状态，可以在任何点暂停和恢复Graph执行，以支持错误恢复、等。
  • 应用场景：对于需要中断和恢复的长时任务非常有用，例如数据分析任务、需要人工审核的流程等。

4. Human-in-the-Loop

  • 功能描述：允许中断Graph的执行，以便人工批准或编辑Agent计划的下一步操作。
  • 应用场景：在需要人工监督和干预的场合，如敏感操作审批、复杂决策支持等。

5. Streaming Support（流支持）

  • 功能描述：支持在节点产生输出时实时流输出（包括Token流）。
  • 应用场景：适用于需要实时数据处理和反馈的场景，如实时数据分析、在线聊天机器人等。

6. Integration with LangChain and LangSmith（与LangChain和LangSmith集成）

  • 功能描述：LangGraph可以与LangChain和LangSmith无缝集成，但并不强制要求它们。
  • 应用场景：增强LangChain和LangSmith的功能，提供更灵活的应用构建方式，特别是在需要复杂流程控制和状态管理的场合。

1-3、安装

pip install -U langgraph

1-4、什么是图？

图（Graph）是数学中的一个基本概念，它由点集合及连接这些点的边集合组成。图主要用于模拟各种实体之间的关系，如网络结构、社会关系、信息流等。以下是图的基本组成部分：
在这里插入图片描述

  • 顶点（Vertex）：图中的基本单元，通常用来表示实体。在社交网络中，每个顶点可能代表一个人；在交通网络中，每个顶点可能代表一个城市或一个交通枢纽。
  • 边（Edge）：连接两个顶点的线，表示顶点之间的关系或连接。边可以有方向（称为有向图），也可以没有方向（称为无向图）。
  • 权重（Weight）：有时边会被赋予一个数值，称为权重，表示两个顶点之间关系的强度或某种度量，如距离、容量、成本等。

图可以根据边的性质分为以下几种：

  • 无向图：边没有方向。
  • 有向图：边有方向，通常用箭头表示。
  • 简单图：没有重复的边和顶点自环（即边的两个端点是不同的顶点，且没有边从一个顶点出发又回到同一个顶点）。
  • 多重图：可以有重复的边或顶点自环。
  • 连通图：在无向图中，任意两个顶点之间都存在路径。

图在计算机科学中有广泛的应用，例如：

  • 网络流问题：如最大流、最小割问题。
  • 路径查找问题：如最短路径、所有路径问题。
  • 社交网络分析：分析社交关系网，识别关键节点等。
  • 推荐系统：通过分析用户之间的关系和偏好来推荐内容。

1-5、为什么选择图？

LangGraph之所以使用“图”这个概念，主要是因为图（Graph）在表达复杂关系和动态流程方面具有天然的优势。以下是使用图概念的一些具体原因：

  • 表达复杂关系：在构建智能体应用时，各组件之间可能存在复杂的关系和交互。图结构可以很好地表示这些关系，包括节点（代表状态或操作）和边（代表转移或关系）。
  • 动态流程管理：智能体在执行任务时，往往需要根据不同的输入或状态动态调整其行为。图结构允许灵活地表示这些动态流程，如循环、分支和并行路径。
  • 可扩展性：图结构易于扩展。随着应用复杂度的增加，可以轻松地在图中添加新的节点和边，而不需要重写整个流程。
  • 可视化：图的可视化特性使得开发者能够直观地理解和调试智能体的行为。通过图形化的表示，可以更快速地识别问题和优化点。
  • 循环和递归：许多智能体应用需要处理循环或递归逻辑，如图结构可以自然地表示这种循环引用和重复过程。
  • 灵活的控制流：与传统的线性流程（如有向无环图DAG）相比，图结构支持更复杂的控制流，包括条件分支和并发执行。
  • 启发式算法和数据流：图算法（如最短路径、网络流等）可以为优化智能体行为提供启发，特别是在处理数据流和资源分配时。

在LangChain的简答链中无法实现的循环场景：

1、代码生成与自我纠正：

  • 场景描述：利用LLM自动生成软件代码，并根据代码执行的结果进行自我反省和修正。
  • LangGraph应用：LangGraph可以创建一个循环流程，首先生成代码，然后测试执行，根据执行结果反馈给LLM，让它重新生成或修正代码，直到达到预期的执行效果。这种循环机制在传统的链式（Chain）结构中难以实现。

2、Web自动化导航：

  • 场景描述：自动在Web上进行导航，例如自动填写表单、点击按钮或从网站上抓取信息。
  • LangGraph应用：LangGraph可以定义一个包含循环的流程，使得智能体能够在进入下一界面时，根据多模态模型的决定来执行不同的操作（如点击、滚动、输入等），直到完成特定任务。这种循环和条件逻辑的运用在LangGraph中得到了很好的支持。

总结来说：LangGraph可以表达更复杂的关系，更灵活，控制更精细，具备循环能力。

1-6、LangGraph应用的简单示例—CRAG（自我改正型RAG）

LangGraph： 是 LangChain 的扩展库，不是独立的框架。它能协调 Chain、Agent 和 Tool 等组件，支持 LLM 循环调用和 Agent 过程的精细化控制。LangGraph 使用状态图（StateGraph）代替了 AgentExecutor 的黑盒调用，通过定义图的节点和边来详细定义基于 LLM 的任务。在任务运行期间，它会维护一个中央状态对象，该对象会根据节点的变化不断更新，其属性可根据需要进行自定义。相比于 AgentExecutor，LangGraph 可以更加精细的进行控制：

在这里插入图片描述

CRAG： 顾名思义，一种RAG的变体，结合了对检索到的文档的自我反思/自我评分。
在这里插入图片描述
图表展示了一个查询处理流程，涉及多个阶段和决策点：

  • Question（提问）：这是整个流程的开始点，用户提出一个问题。
  • Retrieve Node（检索节点）：系统尝试从数据库或索引中检索与问题相关的信息。
  • Grade Node（评分节点）：对检索到的信息进行评估，判断其相关性或准确性。
  • Decision Point（决策点）：根据评分节点的输出，系统会做出是否继续当前路径还是选择替代路径的决定。
    如果没有发现任何无关的文档（“Any doc irrelevant”? “No”），则流程直接跳到“Answer（答案）”节点。
    如果发现了无关的文档（“Any doc irrelevant”? “Yes”），则进入下一个阶段。重查。
  • Re-write Query Node（重写查询节点）：由于检索到的某些文档被认为是不相关的，系统会对原始查询进行重新表述，以便更准确地反映用户的需求。
  • Web Search Node（网页搜索节点）：使用重写的查询在互联网上搜索更多信息。
  • Answer（答案）：最终，系统将生成的答案返回给用户。

节点可以是可调用的函数、工具、Agent、或者是一个可运行的chain。

1-7、LangGraph基础概念

1-7-1、Graphs（图的概念&关键组件&如何构建）

在LangGraph框架中，“Graphs”（图）是核心概念之一，用于图形化 智能体（agents）的工作流程。（即将工作流程建模为图形），主要使用三个关键组件来定义：

  • State: 状态，一个共享的数据结构。
  • Nodes：节点，编辑Agent逻辑的python函数，接收当前状态作为输入，执行一系列计算后，返回更新的状态。
  • Edges：边，基于当前状态决定下一个执行节点。边可以是条件分支，或者固定转换。

通过组合、拼接节点和边，可以创建复杂的工作流程。

Graphs 执行开始时，所有节点都以初始状态开始，当节点完成操作后，它会沿着一条或者多条边向其他节点发送消息，之后，接收方节点执行其函数，将生成的消息传递给下一组节点。直到没有消息传递！
简单说：节点完成操作，边决策下一步干什么。

参数：

  • StateGraph：状态图，用于将用户定义的对象参数化。
  • MessageGraph：消息图，除了聊天机器人外基本不使用。

构建图： 首先需要定义state，之后需要添加各个节点和边，最后就可以编译图了。（对图结构的一些基本检查，确保图没有孤立节点，另外还可以指定一些运行时的参数）。调用以下方法来编译图：

graph = graph_builder.compile(...)

1-7-2、State（状态）

State：

  1. 定义： 状态是Graph中的一个共享数据结构，它存储了所有与当前执行上下文相关的信息。
  2. 数据结构： 状态可以是任何Python类型，通常使用TypedDict或PydanticBaseModel。TypedDict是一个Python字典，它允许对字典键和值的类型进行注解。
  3. 作用： 状态用于存储和传递应用程序的数据，使得节点可以基于这些数据执行操作。它提供了节点之间的通信机制，因为每个节点都可以读取前一个节点更新的状态，并在此基础上进行操作。
  4. 管理： 状态的管理是自动的。当一个节点执行并返回一个更新后的状态时，LangGraph框架会确保这个新状态被传递到下一个节点。
  5. 生命周期： 状态的生命周期与图的执行周期相匹配，它从图的初始状态开始，并在图的每个节点执行时更新，直到图执行结束。

1-7-3、Annotated（数据类型）

Annotated作用：

  1. 元数据添加：Annotated允许开发者在类型提示中添加额外的信息，这些信息可以被类型检查器、框架或其他工具使用。
  2. 类型提示增强：它提供了一种方式来增强现有的类型提示，而不需要创建新的类型。
  3. 代码文档：Annotated可以作为一种文档形式，提供关于变量、函数参数或返回值的额外信息。

用法1： DistanceInCm是一个带注释的整数类型。注释 “Units: cm” 说明了这个整数代表的是以厘米为单位的距离。注释可以用来作为文档，说明变量的用途或期望的值。

from typing import Annotated
from typing_extensions import Annotated  # 如果标准库中没有Annotated

# 定义一个带注释的整数类型
# 这里的 "Units: cm" 是一个注释，它不会改变类型的行为
DistanceInCm = Annotated[int, "Units: cm"]

def measure_distance(distance: DistanceInCm) -> DistanceInCm:
    # 这里我们假设函数会测量距离，并返回以厘米为单位的距离
    # 注意：函数的实现并不关心注释 "Units: cm"
    return distance

# 使用带注释的类型
distance: DistanceInCm = 10  # 10 厘米
new_distance = measure_distance(distance)

用法2： messages 变量，其类型被注解为 Annotated[list, add_messages]。list 表示 messages 键的值应该是一个列表。add_messages 是一个函数，它在 Annotated 注解中使用，提供了关于如何更新状态字典中 messages 的额外信息。

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    # Messages have the type "list". The `add_messages` function
    # in the annotation defines how this state key should be updated
    # (in this case, it appends messages to the list, rather than overwriting them)
    messages: Annotated[list, add_messages]

1-7-4、Node（节点）

Node： 在LangGraph框架中，节点（Nodes）是Python函数，它们编码了智能体（agents）的逻辑。其中第一个位置参数是State（名称）。第二个位置参数是config（Node对应的处理逻辑）。

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

builder = StateGraph(dict)


def my_node(state: dict, config: RunnableConfig):
    print("In node: ", config["configurable"]["user_id"])
    return {"results": f"Hello, {state['input']}!"}


# The second argument is optional
def my_other_node(state: dict):
    return state


builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)
...

Start节点： 特殊节点，表示用户输入发送到Graph的节点。该节点的主要目的是确定首先应该调用哪些节点。

from langgraph.graph import START

graph.add_edge(START, "node_a")

End节点： 特殊节点，确定哪条边完成后，没有后续操作。

from langgraph.graph import END

graph.add_edge("node_a", END)

1-7-5、Edge（边）

在LangGraph框架中，边（Edges）是用于连接节点的对象，它们定义了节点之间的转换逻辑。每条边都连接两个节点：一个源节点和一个目标节点。边的主要功能是确定何时应该从源节点跳转到目标节点。

  • Normal Edges：正常边，直接从一个节点转到下一个节点。
  • Conditional Edges：调用一个函数以确定接下来要转到哪个节点。
# 正常边，直接从节点A跳转到节点B
graph.add_edge("node_a", "node_b")

# 条件边，从节点A选择性的跳转到下一条边，routing_function为跳转的逻辑方法。
graph.add_conditional_edges("node_a", routing_function)

1-7-6、Command

概念： Command可以很方便的既进行走向控制，又可以更新状态。个人理解，代码更加简洁，省去使用条件边的流程。

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

动态控制： 类似条件边

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    if state["foo"] == "bar":
        return Command(update={"foo": "baz"}, goto="my_other_node")

二、Multi-agent supervisor构建

2-1、定义

Multi-agent network： 处理复杂任务的一种有效方法是，分而治之，即为每个任务创建一个Agent。

在这里插入图片描述

2-2、创建工具

tavily 搜索API申请地址: https://docs.tavily.com/docs/rest-api/api-reference

pip install -U langchain-community tavily-python

TavilySearchResults参数介绍:

  • max_results：最大返回搜索数量
  • include_answer：是否包含答案
  • include_images： 是否包含图片

简易Demo：

import os 

os.environ["TAVILY_API_KEY"] = ""

from langchain_community.tools import TavilySearchResults

tool = TavilySearchResults(
    max_results=5,
    include_answer=True,
    include_raw_content=True,
    include_images=True,
    # search_depth="advanced",
    # include_domains = []
    # exclude_domains = []
)
tools = [tool]
tool.invoke({'query': '谁是世界上最美丽的女人？'})

创建搜索工具&Python执行工具：


from typing import Annotated

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL

tavily_tool = TavilySearchResults(max_results=5)

# Warning: This executes code locally, which can be unsafe when not sandboxed

repl = PythonREPL()


@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """Use this to execute python code. If you want to see the output of a value,
    you should print it out with `print(...)`. This is visible to the user."""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"Failed to execute. Error: {repr(e)}"
    result_str = f"Successfully executed:\n\`\`\`python\n{code}\n\`\`\`\nStdout: {result}"
    return (
        result_str + "\n\nIf you have completed all tasks, respond with FINAL ANSWER."
    )

2-3、构建Agent节点

智能体初始化：

  • research_agent：负责研究英国GDP数据的智能体。它使用TavilySearchResults工具进行搜索。
  • chart_agent：负责生成图表的智能体。它使用python_repl_tool工具，该工具允许执行Python代码来创建图表。

系统提示： make_system_prompt函数为每个智能体创建了一个提示，指定了它们的角色以及合作的同事。

节点： 以research_node为例。研究智能体，它处理状态并决定下一步行动，是将任务传递给图表生成器还是完成任务。

def make_system_prompt(suffix: str) -> str:
    return (
        "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."
        f"\n{suffix}"
    )


from typing import Literal

from langchain_core.messages import BaseMessage, HumanMessage
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from langgraph.graph import MessagesState, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI
import os


llm = ChatOpenAI(
    model="qwen-max",
    temperature=0,
    max_tokens=1024,
    timeout=None,
    max_retries=2,
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)


# 决策下一步到哪个节点，判断FINAL ANSWER是否在最后一条消息中，如果在的话，就返回END。
def get_next_node(last_message: BaseMessage, goto: str):
    if "FINAL ANSWER" in last_message.content:
        # Any agent decided the work is done
        return END
    return goto


# Research agent and node
research_agent = create_react_agent(
    llm,
    tools=[tavily_tool],
    state_modifier=make_system_prompt(
        "You can only do research. You are working with a chart generator colleague."
    ),
)


# -> Command[Literal["chart_generator", END]]： 限制返回数据类型为"chart_generator", END，提高程序的可读性以及可靠性。
def research_node(
    state: MessagesState,
) -> Command[Literal["chart_generator", END]]:
    result = research_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "chart_generator")
    # wrap in a human message, as not all providers allow
    # AI message at the last position of the input messages list
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="researcher"
    )
    return Command(
        update={
            # share internal message history of research agent with other agents
            "messages": result["messages"],
        },
        goto=goto,
    )


# Chart generator agent and node
# NOTE: THIS PERFORMS ARBITRARY CODE EXECUTION, WHICH CAN BE UNSAFE WHEN NOT SANDBOXED
chart_agent = create_react_agent(
    llm,
    [python_repl_tool],
    state_modifier=make_system_prompt(
        "You can only generate charts. You are working with a researcher colleague."
    ),
)


def chart_node(state: MessagesState) -> Command[Literal["researcher", END]]:
    result = chart_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "researcher")
    # wrap in a human message, as not all providers allow
    # AI message at the last position of the input messages list
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="chart_generator"
    )
    return Command(
        update={
            # share internal message history of chart agent with other agents
            "messages": result["messages"],
        },
        goto=goto,
    )

2-4、Graph构造

  • create_react_agent：创建React范式的Agent。
from langgraph.graph import StateGraph, START

workflow = StateGraph(MessagesState)
workflow.add_node("researcher", research_node)
workflow.add_node("chart_generator", chart_node)

workflow.add_edge(START, "researcher")
graph = workflow.compile()

调用顺序如下所示：
在这里插入图片描述

2-5、调用

任务： 首先收集过去五年英国GDP的数据（研究者节点），然后根据这些数据生成一张折线图（图表节点）。

events = graph.stream(
    {
        "messages": [
            (
                "user",
                "First, get the UK's GDP over the past 5 years, then make a line chart of it. "
                "Once you make the chart, finish.",
            )
        ],
    },
    # Maximum number of steps to take in the graph
    {"recursion_limit": 150},
)
for s in events:
    print(s)
    print("----")

输出：

在这里插入图片描述

参考文章：
LlamaIndex 官方文档
langgraph官方教程
langgraph操作指南
langgraph概念指南
langgraph API 参考
langgraph 词汇表
langgraph 快速入门
彻底搞懂LangGraph【1】：构建复杂智能体应用的LangChain新利器
LangChain 79 LangGraph 从入门到精通一

总结

心累😅


http://www.kler.cn/a/472347.html

相关文章：

  • go 1.23.4安装
  • 常用的数据引擎及其特点
  • 高阶知识库搭建实战五、（向量数据库Milvus安装）
  • 虚幻（UE）资源网站
  • gaussdb怎么查询一个表所在的表空间的总大小和可用大小，用GB为单位表示？
  • 【每日学点鸿蒙知识】关于热修复、图片预览、多个@State刷新性能问题等
  • 【网络安全技术与应用】（选修）实验2 用Wireshark分析典型TCP/IP体系中的协议
  • Web前端ui框架
  • LLM 训练中存储哪些矩阵：权重矩阵，梯度矩阵，优化器状态
  • javaCV音频剪切
  • 我的AI工具箱Tauri版-ZoomImageFlux图像缩放
  • 【网络安全 | 漏洞挖掘】HubSpot 全账户接管（万字详析）
  • Linux 安装 Mosquitto 及 SpringBoot 整合
  • 用JAVA编写一个简单的小游戏
  • pdf在页面中预览的方法
  • 数据标注「orc」
  • 路由组件与一般组件的区别
  • UDP_TCP
  • 应急指挥与调度子模块示例
  • 解密Navicat密码（Java）