LangChain-LangGraph框架 应用实例

LangGraph是LangChain的高级库，为大型语言模型（LLM）带来循环计算能力。

LangGraph建立在LCEL之上，允许高效地协调应用程序组件，同时保持简洁和可读的代码。它包括内置持久性、对循环的支持，并优先考虑可控性。 如果LCEL对于更大或更复杂的链变得笨重，它们可能会受益于LangGraph的实现。

状态（State） 状态是一个共享的数据结构，通常是一个TypedDict或者Pydantic的BaseModel类型。

节点（Nodes） 节点是一个Python函数，接受一个State作为输入，经过内部计算后，返回更新后的State。

边（Edges） 边也是一个Python函数，基于当前State，决定下一步执行哪个/哪一些节点。

案例1

以下是 是机器人对话，使用LangGraph框架构建State、Nodes、Edges，创建一个流程图

先安装包：langgraph

步骤如下：

1.定义一个状态类型：MyState

messages 代表状态中保存数据的key list代表数据类型        
add_messages函数用于更新列表数据

class MyState(TypedDict):
    messages:Annotated[list,add_messages]

2.创建一个流程图

graph = StateGraph(MyState)

3.准备一个node节点，大模型节点

llm = ChatOpenAI(
    temperature=1,
    model='deepseek-r1',
    api_key='sk-*****',
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

4.创建一个机器人节点函数，以当前state作为输入，并返回一个包含更新后的messages

然后使用add_node方法，将机器人节点加入到流程图中，该方法的

第一个参数是唯一节点的名称

第二个参数是当前节点被使用时将调用的函数或对象

def chatbot(state:MyState):
    return {'messages':[llm.invoke(state['messages'])]}

graph.add_node('chatbot',chatbot)

5.设置边，构建一张图

graph.add_edge(START,'chatbot') #表示流程从 START--->chatbot
graph.add_edge('chatbot',END) #表示流程从 chatbot--->END

graph = graph.compile()

6.最后，为了让机器人可以一直和用户对话，循环调用这个流程图，需要创建方法loop_graph_invoke

def loop_graph_invoke(user_input:str):
    for chunk in graph.stream({'messages':[('user',user_input)]}):
        for value in chunk.values():
            print('AI机器人：',value['messages'][-1].content)

while True:
    try:
        user_input = input('用户：')
        if(user_input.lower() in ['q','exit','quit']):
            print('对话结束，欢迎下次使用！')
            break
        else:
            loop_graph_invoke(user_input)
    except Exception as e:
        print(e)

完整代码：

from typing import Annotated
from langchain_openai import ChatOpenAI
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from typing_extensions import TypedDict



class MyState(TypedDict):
    #messages 代表状态中保存数据的key list代表数据类型，add_messages函数用于更新列表数据
    messages:Annotated[list,add_messages]

graph = StateGraph(MyState)

llm = ChatOpenAI(
    temperature=1,
    model='deepseek-r1',
    api_key='sk-*****',
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

def chatbot(state:MyState):
    return {'messages':[llm.invoke(state['messages'])]}


graph.add_node('chatbot',chatbot)


#设置边
graph.add_edge(START,'chatbot') #表示流程从 START--->chatbot
graph.add_edge('chatbot',END) #表示流程从 chatbot--->END

#构建一张图
graph = graph.compile()

#把graph变成一张图


def loop_graph_invoke(user_input:str):
    """循环调用这个流程图，让AI可以一直和用户对话"""
    for chunk in graph.stream({'messages':[('user',user_input)]}):
        for value in chunk.values():
            print('AI机器人：',value['messages'][-1].content)

while True:
    try:
        user_input = input('用户：')
        if(user_input.lower() in ['q','exit','quit']):
            print('对话结束，欢迎下次使用！')
            break
        else:
            loop_graph_invoke(user_input)
    except Exception as e:
        print(e)

运行结果：

案例2

基于上面的机器人案例，以下是结合工具Tavily，实现实时搜索，流程图如下

首先需要添加工具，然后和大模型进行绑定

os.environ['TAVILY_API_KEY'] = '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'
search_