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

langchain系列（九）- LangGraph 子图详解

news 来源：原创 2025/6/8 14:00:29

一、导读

环境：OpenEuler、Windows 11、WSL 2、Python 3.12.3 langchain 0.3 langgraph 0.2

背景：前期忙碌的开发阶段结束，需要沉淀自己的应用知识，过一遍LangGraph

时间：20250307

说明：技术梳理，LangGraph 的多代理（多智能体）基于子图实现，此处对子图进行说明，案例基于官方文档进行部分的修改

官方文档地址：LangGraph 子图实现

二、原理说明

1、简介

子图允许您构建具有多个组件的复杂系统，这些组件本身是图。使用子图的常见用例是构建多代理系统。

在添加子图时，主要问题是父图和子图如何进行通信，即它们如何在图执行期间相互传递状态。这里有两种情况：

父图和子图共享schema keys。在这种情况下，您可以添加一个编译后的子图节点

父图和子图具有不同的schemas。在这种情况下，您必须添加一个调用子图的节点函数：当父图和子图具有不同的状态模式时，这在调用子图之前或之后需要转换状态时很有用

2、子图图示

3、使用说明

一种常见的情况是父图和子图通过共享状态键（通道）进行通信。例如，在多代理系统中，代理通常通过共享的消息键进行通信。

如果你的子图与父图共享状态键，你可以按照以下步骤将其添加到父图中：

定义子图工作流（如下例中的subgraph_builder）并进行编译

在定义父图工作流时，将编译后的子图传递给.add_node方法

三、代码实现

1、实现功能

开始节点进行意图分类，如果天气相关则走天气相关的子图；否则走智能助手节点。

2、Graph 图示

3、代码实现

from langgraph.graph import StateGraph, END, START, MessagesState
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from typing import Literal
from langgraph.checkpoint.memory import MemorySaver

@tool
def get_weather(city: str):
    """获取天气信息"""
    return f"{city}，晴空万里，阳光明媚!"

# 添加记忆存储
memory = MemorySaver()

# 配置信息，用于记录对话记录
config = {"configurable": {"thread_id": "1"}}


# 指定大模型的API Key 等相关信息
llm = ChatOpenAI(
    base_url="https://lxxxxx.enovo.com/v1/", 
    api_key="sxxxxxxxwW",
    model_name="qwen2.5-instruct"
    )


# 绑定工具
model = raw_model.bind_tools([get_weather])

# 子图的state
class SubGraphState(MessagesState):
    city: str

# 识别地点（城市）
def model_node(state: SubGraphState):
    system_message = """用户的问题是某个地方的天气问题，请辨别具体城市名称，并输出城市名称。"""
    messages = [{"role": "system", "content": system_message}] + state["messages"]
    result = raw_model.invoke(messages)
    return {"city": result.content}

# 天气工具节点
def weather_node(state: SubGraphState):
    result = get_weather.invoke({"city": state["city"]})
    return {"messages": [{"role": "assistant", "content": result}]}

# 子图的添加节点、边以及编译图
subgraph = StateGraph(SubGraphState)
subgraph.add_node(model_node)
subgraph.add_node(weather_node)
subgraph.add_edge(START, "model_node")
subgraph.add_edge("model_node", "weather_node")
subgraph.add_edge("weather_node", END)
subgraph = subgraph.compile()

# 意图分类state
class RouterState(MessagesState):
    route: Literal["天气", "其他"]

# 意图分类节点
def router_node(state: RouterState):
    system_message = """用户输入与天气相关则输出"天气"，与天气无关返回"其他”"""
    messages = [{"role": "system", "content": system_message}] + state["messages"]
    route = raw_model.invoke(messages)
    return {"route": route.content}

# 智能助手节点
def normal_llm_node(state: RouterState):
    response = raw_model.invoke(state["messages"])
    return {"messages": [response]}

# 选择边函数
def route_after_prediction(state: RouterState):
    if state["route"] == "天气":
        return "weather_graph"
    else:
        return "normal_llm_node"

# 父图的添加边、节点以及编译父图
graph = StateGraph(RouterState)
graph.add_node(router_node)
graph.add_node(normal_llm_node)
# 将子图作为一个节点添加到父图中
graph.add_node("weather_graph", subgraph)
graph.add_edge(START, "router_node")
graph.add_conditional_edges("router_node", route_after_prediction)
graph.add_edge("normal_llm_node", END)
graph.add_edge("weather_graph", END)
graph = graph.compile(checkpointer=memory)


# 以流式调用图
def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}, config=config, stream_mode="updates"):
        if event.get("router_node"):
            continue
        else:
            for value in event.values():
                print("Assistant:", value["messages"][-1].content)

# 设置聊天退出方法
while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break

        stream_graph_updates(user_input)
    except Exception as e:
        print(e)
        user_input = "What do you know about LangGraph?"
        print("User: " + user_input)
        stream_graph_updates(user_input)
        break