langgraph使用

langgraph使用

起男 14 2025-05-18

langgraph使用

简单上手

状态

from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: str

节点

def node_1(state: State):
    print("node 1")
    return {"graph_state": state["graph_state"]+"我很"}
def node_2(state: State):
    print("node 2")
    return {"graph_state": state["graph_state"]+"高兴！"}
def node_3(state: State):
    print("node 3")
    return {"graph_state": state["graph_state"]+"悲伤！"}

条件边

from typing import Literal
import random

def decide_mood(state: State) -> Literal["node_2","node_3"]:
    # 通常使用状态来决定下一个要访问的节点
    #user_input = state["graph_state"]
    #这里随机访问
    if random.random() < 0.5:
        return "node_2"
    return "node_3"

定义图

from langgraph.graph import StateGraph,START,END

builder = StateGraph(State)
#设置节点
builder.add_node(node_1)
builder.add_node(node_2)
builder.add_node(node_3)
#添加边
builder.add_edge(START,"node_1")
#条件边
builder.add_conditional_edges("node_1",decide_mood)
builder.add_edge("node_2",END)
builder.add_edge("node_3",END)

graph = builder.compile()

测试

s = graph.invoke({"graph_state":"你好"})
print(s)

整合大模型和Tools

大模型

from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    api_key=os.environ.get("ALI_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-max"
)

定义工具并绑定

from langchain_core.tools import tool
@tool
def multiply(a:int,b:int)->int:
    """计算a乘b"""
    return a*b
#绑定
llm_with_tools = llm.bind_tools([multiply])

状态

from typing_extensions import TypedDict
from typing import Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage

class MessageState(TypedDict):
    # 每次更新进行追加
    messages : Annotated[list[AnyMessage],add_messages]

节点

def tool_calling_llm(state : MessageState):
    return {"messages":[llm_with_tools.invoke(state["messages"])]}

构建图

builder = StateGraph(MessageState)
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_edge(START,"tool_calling_llm")
builder.add_edge("tool_calling_llm",END)
graph = builder.compile()

测试

messages = graph.invoke({"messages":HumanMessage("2*3等于多少")})
for m in messages["messages"]:
    m.pretty_print()

此时是无法自动调用tools的

自动调用tools

构建图时将

from langgraph.prebuilt import ToolNode,tools_condition

builder = StateGraph(MessageState)
builder.add_node("tool_calling_llm", tool_calling_llm)
#将工具封装成节点
builder.add_node("tools",ToolNode([multiply]))
builder.add_edge(START,"tool_calling_llm")
#如果llm的结果是工具调用则路由到工具，否则路由的END
builder.add_conditional_edges("tool_calling_llm",tools_condition)
builder.add_edge("tools",END)
graph = builder.compile()

ReAct架构

定义工具并绑定

@tool
def multiply(a:int,b:int)->int:
    """计算a乘b"""
    return a*b
@tool
def add(a:int,b:int)->int:
    """计算a加b"""
    return a+b
@tool
def divide(a:int,b:int)->float:
    """计算a除b"""
    return a/b
#绑定
tools = [add,multiply,divide]
# parallel_tool_calls关闭工具并行
llm_with_tools = llm.bind_tools(tools,parallel_tool_calls=False)

定义状态和节点

# 提示词
sys_msg = SystemMessage("你是一个乐于助人的助手，负责对一组输入进行算数运算")
#状态
class MessageState(TypedDict):
    # 每次更新进行追加
    messages : Annotated[list[AnyMessage],add_messages]
# 节点
def assistant(state : MessageState):
    return {"messages":[llm_with_tools.invoke([sys_msg]+state["messages"])]}

定义图

builder = StateGraph(MessageState)
# 定义节点
builder.add_node("assistant", assistant)
builder.add_node("tools",ToolNode(tools))
#定义边
builder.add_edge(START,"assistant")
builder.add_conditional_edges("assistant",tools_condition)
# 循环工具到llm的边
builder.add_edge("tools","assistant")
graph = builder.compile()

测试

messages = graph.invoke({"messages":HumanMessage("3加4再乘2最后除以3是多少")})
for m in messages["messages"]:
    m.pretty_print()

实现记忆

在构建图时，设置checkpointer检查点以实现记忆功能

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

测试

#定义一个线程id 这里将记忆点写死为1
config = {"configurable":{"thread_id":"1"}}
#调用
messages1 = graph.invoke({"messages":HumanMessage("3加4是多少")},config)
for m in messages1["messages"]:
    m.pretty_print()
messages2 = graph.invoke({"messages":HumanMessage("再乘2是多少")},config)
for m in messages2["messages"]:
    m.pretty_print()
llm python