【LangChain高级系列】LangGraph第一课

news2025/5/10 12:31:54

前言

我们今天直接通过一个langgraph的基础案例，来深入探索langgraph的核心概念和工作原理。

基本认识

LangGraph是一个用于构建具有LLMs的有状态、多角色应用程序的库，用于创建代理和多代理工作流。与其他LLM框架相比，它提供了以下核心优势：循环、可控性、持久性。

LangGraph允许您定义涉及循环的流程，这对于大多数代理架构至关重要。作为一种非常底层的框架，它提供了对应用程序的流程和状态的精细控制，这对创建可靠的代理至关重要。

此外,LangGraph包含内置的持久性，可以实现高级的"人机交互"和内存功能。LangGraph是LangChain的高级库，为大型语言模型(LLM)带来循环计算能力。它超越了LangChain的线性工作流，通过循环支持复杂的任务处理。

核心概念

状态（State）：维护计算过程中的上下文,实现基于累积数据的动态决策。
节点（Node）：代表计算步骤,执行特定任务,可定制以适应不同工作流。
边（Edge）：连接节点,定义计算流程,支持条件逻辑,实现复杂工作流

Langgraph执行细节

from typing import Literal
from langchain.tools import tool
from langgraph.graph import END, MessagesState, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
from langchain_deepseek import ChatDeepSeek
import os
# 执行轨迹上报
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_3df2ddbe15e308db82d935249e9e2_7668b99099"     
os.environ["LANGCHAIN_PROJECT"] = "langgraph_base"
os.environ["LANGSMITH_ENDPOINT"] = "https://api.smith.langchain.com"

# 工具函数，用于查询天气
@tool(description="查询天气")
def query_weather(words: str):
    if "上海" in words.lower() or "shanghai" in words.lower():
        return "天气晴朗，适合出行"
    else:
        return "天晴欠佳，不宜出行"

# 工具函数，用于查询天气
@tool(description="查询交通")
def query_thransport(words: str):
    if "交通" in words.lower() or "transport" in words.lower():
        return "交通通畅，可自驾"
    else:
        return "交通拥堵，不宜出行"

# 将查询天气的工具函数添加到工具列表中
tools = [query_weather, query_thransport]

# 将工具列表转换为工具节点，便于langgraph调用工具
tool_nodes = ToolNode(tools)

llm = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    api_key="sk-37c6b1c6517141e644c5ba1c81782",
).bind_tools(tools) # 将工具绑定到llm模型上，以便模型可以调用工具函数

def action_router(state: MessagesState) -> Literal["tools", END]: # 定义动作路由函数，用于根据当前状态选择下一步动作
    messages = state["messages"] # 获取当前消息列表，用于判断最后一条消息的类型，以便选择下一步动作
    last_message = messages[-1] # 获取最后一条消息，用于判断最后一条消息的类型，以便选择下一步动作
    if last_message.tool_calls: # 如果最后一条消息包含工具调用，则选择工具节点
        return "tools"
    else: # 否则结束
        return END

def call_llm(state: MessagesState) -> MessagesState: # 定义调用llm模型的函数，用于生成回复消息
    messages = state["messages"] # 获取当前消息列表，用于生成回复消息
    response = llm.invoke(messages) # 调用llm模型生成回复消息
    return {"messages": [response]} # 返回回复消息

workflow = StateGraph(MessagesState) # 创建状态图，用于管理消息状态

workflow.add_node("agent", call_llm) # 将工具节点添加到状态图中，以便工具节点可以被调用
workflow.add_node("tools", tool_nodes) # 将调用llm模型的函数添加到状态图中，以便llm模型可以被调用

workflow.set_entry_point("agent") # 设置入口点，即从工具节点开始

workflow.add_conditional_edges( # 添加条件边，即根据当前状态选择下一步动作
    "agent", # 从工具节点开始
    action_router # 选择下一步动作的函数
) # 结束添加条件边

workflow.add_edge("tools", "agent") # 将工具节点和调用llm模型的函数连接起来，以便工具节点可以调用llm模型生成回复消息

app = workflow.compile(checkpointer=MemorySaver()) # 编译状态图，以便可以被调用

final_state = app.invoke(
    {"messages": [{"role": "user", "content": "交通状态如何？"}]}, 
    config={"configurable": {"thread_id": "123"}} # 调用状态图，生成回复消息，传入配置参数，以便可以被调用
) # 调用状态图，生成回复消息

print(final_state) # 打印回复消息

#将生成的图片保存到文件
graph_png = app.get_graph().draw_mermaid_png()
with open("langgraph_base.png", "wb") as f:
    f.write(graph_png)

当我们输入：交通状况如何？大模型会选择query_transport工具

当我们输入：上海天气如何？大模型会选择query_weather工具

上图中，_start_和_end_是虚拟节点，其中虚线是条件边，实线是一定会执行的，也就是说执行完tools后，一定会再次调用agent逻辑。

Langsmith监控分析

1、上图展示的是执行轨迹的瀑布流模式的展现形式

2、从上图看，以此调用一共耗时12.48秒

3、然后是agent节点的执行，耗时6.88秒，agent会调用deepseek，我们看看deepseek返回的数据：

{
"generations": [
[
{
"text": "",
"generation_info": {
"finish_reason": "tool_calls",
"logprobs": null
},
"type": "ChatGeneration",
"message": {
"lc": 1,
"type": "constructor",
"id": [
"langchain",
"schema",
"messages",
"AIMessage"
],
"kwargs": {
"content": "",
"additional_kwargs": {
"tool_calls": [
{
"id": "call_0_3d3d77c8-919e-42c1-a402-b4f87bf248be",
"function": {
"arguments": "{\"words\":\"上海的天气\"}",
"name": "query"
},
"type": "function",
"index": 0
}
],
"refusal": null
},
"response_metadata": {
"token_usage": {
"completion_tokens": 18,
"prompt_tokens": 98,
"total_tokens": 116,
"completion_tokens_details": null,
"prompt_tokens_details": {
"audio_tokens": null,
"cached_tokens": 64
},
"prompt_cache_hit_tokens": 64,
"prompt_cache_miss_tokens": 34
},
"model_name": "deepseek-chat",
"system_fingerprint": "fp_8802369eaa_prod0425fp8",
"id": "4e9ca352-44d5-4612-a02f-da0340c9dffc",
"finish_reason": "tool_calls",
"logprobs": null
},
"type": "ai",
"id": "run--b6cf271e-a3b3-4d6a-805c-e15a921be943-0",
"tool_calls": [
{
"name": "query",
"args": {
"words": "上海的天气"
},
"id": "call_0_3d3d77c8-919e-42c1-a402-b4f87bf248be",
"type": "tool_call"
}
],
"usage_metadata": {
"input_tokens": 98,
"output_tokens": 18,
"total_tokens": 116,
"input_token_details": {
"cache_read": 64
},
"output_token_details": {}
},
"invalid_tool_calls": []
}
}
}
]
],
"llm_output": {
"token_usage": {
"completion_tokens": 18,
"prompt_tokens": 98,
"total_tokens": 116,
"completion_tokens_details": null,
"prompt_tokens_details": {
"audio_tokens": null,
"cached_tokens": 64
},
"prompt_cache_hit_tokens": 64,
"prompt_cache_miss_tokens": 34
},
"model_name": "deepseek-chat",
"system_fingerprint": "fp_8802369eaa_prod0425fp8",
"id": "4e9ca352-44d5-4612-a02f-da0340c9dffc"
},
"run": null,
"type": "LLMResult"
}