【LangGraph】如何使用 LangGraph

米落枫 · 发表于昨天 22:47

作者：彬彬侠
本文介绍了如何使用 LangGraph 来构建和管理基于大型语言模型（LLM）的复杂工作流。从安装开始，逐步讲解核心步骤、关键概念、代码示例，以及如何在实际场景中应用 LangGraph。

1. 安装和环境准备

在使用 LangGraph 之前，你需要设置开发环境并安装必要的依赖。
1.1 安装 LangGraph 和 LangChain

LangGraph 是 LangChain 生态的一部分，因此需要安装 LangChain 和 LangGraph。

pip install langchain langgraph

复制代码

如果你计划使用特定的 LLM（如 OpenAI 的 GPT 模型），还需要安装对应的客户端库。例如：

pip install langchain-openai # 用于 OpenAI API

复制代码

1.2 配置 API 密钥

如果使用外部 LLM 服务（如 OpenAI、Anthropic），需要配置 API 密钥。可以通过环境变量设置：

exportOPENAI_API_KEY="your-api-key"

复制代码

或者在代码中手动设置（不推荐用于生产环境）：

import os
os.environ["OPENAI_API_KEY"]="your-api-key"

复制代码

1.3 验证安装

运行以下代码，检查是否正确安装：

from langgraph.graph import StateGraph
print("LangGraph installed successfully!")

复制代码

2. LangGraph 的核心概念回顾

在开始使用之前，快速回顾 LangGraph 的核心概念（详细介绍见前文）：

状态（State）

节点（Node）

边（Edge）

图（Graph）

LangGraph 的执行逻辑是：输入初始状态 → 图按节点和边定义的顺序执行 → 不断更新状态 → 直到达到终止条件。

3. 构建一个 LangGraph 工作流：分步指南

以下是使用 LangGraph 构建工作流的标准步骤：
3.1 定义状态

状态是工作流的核心，用于存储和传递信息。可以使用 Python 的 TypedDict 或自定义类来定义状态。
示例状态定义：

from typing import TypedDict
classState(TypedDict):input:str# 用户输入
output:str# 处理结果

复制代码

3.2 创建节点

节点是执行任务的函数，接收状态并返回更新后的状态。每个节点是一个 Python 函数。
示例节点：

defprocess_input(state: State)-> State:
state["output"]=f"处理输入: {state['input']}"return state
deffinalize_output(state: State)-> State:
state["output"]+=" -> 已完成"return state

复制代码

3.3 定义边

边指定节点之间的执行顺序。可以使用：

3.4 构建图

使用 StateGraph 类创建图，添加节点、边，并设置入口和出口。

from langgraph.graph import StateGraph, START, END
# 创建图
workflow = StateGraph(State)# 添加节点
workflow.add_node("process_input", process_input)
workflow.add_node("finalize_output", finalize_output)# 添加边
workflow.add_edge("process_input","finalize_output")
workflow.add_edge(START,"process_input")# 设置入口
workflow.add_edge("finalize_output", END)# 设置出口# 编译图
graph = workflow.compile()

复制代码

3.5 执行工作流

使用 invoke 方法运行图，传入初始状态。

result = graph.invoke({"input":"Hello, LangGraph!","output":""})print(result)

复制代码

4. 代码示例：简单工作流

以下是一个完整的简单工作流示例，展示如何处理用户输入并生成输出。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 定义状态classState(TypedDict):input:str
output:str# 定义节点defprocess_input(state: State)-> State:
state["output"]=f"处理输入: {state['input']}"return state
deffinalize_output(state: State)-> State:
state["output"]+=" -> 已完成"return state
# 创建图
workflow = StateGraph(State)
workflow.add_node("process_input", process_input)
workflow.add_node("finalize_output", finalize_output)
workflow.add_edge(START,"process_input")
workflow.add_edge("process_input","finalize_output")
workflow.add_edge("finalize_output", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"input":"Hello, LangGraph!","output":""})print(result)

复制代码

输出：

{'input':'Hello, LangGraph!','output':'处理输入: Hello, LangGraph! -> 已完成'}

复制代码

这个示例展示了：

5. 高级功能：条件边、循环和工具调用

LangGraph 的强大之处在于支持动态逻辑、循环和工具调用。以下是实现这些功能的方法。
5.1 条件边

条件边允许根据状态动态选择下一个节点。使用 add_conditional_edges 定义条件逻辑。
示例：根据输入内容决定下一步：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
classState(TypedDict):input:str
output:str
needs_processing:booldefcheck_input(state: State)-> State:
state["needs_processing"]=len(state["input"])>5# 输入长度大于5需要处理return state
defprocess_input(state: State)-> State:
state["output"]=f"处理输入: {state['input']}"return state
defskip_processing(state: State)-> State:
state["output"]="输入太短，无需处理"return state
# 条件函数defroute(state: State)->str:return"process_input"if state["needs_processing"]else"skip_processing"# 创建图
workflow = StateGraph(State)
workflow.add_node("check_input", check_input)
workflow.add_node("process_input", process_input)
workflow.add_node("skip_processing", skip_processing)
workflow.add_edge(START,"check_input")
workflow.add_conditional_edges("check_input", route,{"process_input":"process_input","skip_processing":"skip_processing"})
workflow.add_edge("process_input", END)
workflow.add_edge("skip_processing", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"input":"Hi","output":"","needs_processing":False})print(result)

复制代码

输出（输入 “Hi”）：

{'input':'Hi','output':'输入太短，无需处理','needs_processing':False}

复制代码

输出（输入 “Hello, LangGraph!”）：

{'input':'Hello, LangGraph!','output':'处理输入: Hello, LangGraph!','needs_processing':True}

复制代码

5.2 循环

LangGraph 支持循环，适合需要反复执行的场景（如 agent 多次调用工具）。可以通过条件边返回到之前的节点。
示例：循环直到满足条件：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
classState(TypedDict):
counter:int
output:strdefincrement(state: State)-> State:
state["counter"]+=1
state["output"]=f"当前计数: {state['counter']}"return state
defshould_continue(state: State)->str:return"increment"if state["counter"]<3else END
# 创建图
workflow = StateGraph(State)
workflow.add_node("increment", increment)
workflow.add_edge(START,"increment")
workflow.add_conditional_edges("increment", should_continue,{"increment":"increment",
END: END
})# 编译和运行
graph = workflow.compile()
result = graph.invoke({"counter":0,"output":""})print(result)

复制代码

输出：

{'counter':3,'output':'当前计数: 3'}

复制代码

5.3 工具调用

LangGraph 可以与 LangChain 的工具集成，动态调用外部 API 或函数。
示例：使用 LLM 和搜索工具：

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 定义工具@tooldefsearch(query:str)->str:"""模拟搜索工具"""returnf"搜索结果 for {query}: 找到了一些信息！"# 定义状态classState(TypedDict):input:str
output:str
needs_search:bool# 节点：调用 LLM 判断是否需要搜索defcall_llm(state: State)-> State:
llm = ChatOpenAI(model="gpt-4o-mini")
prompt =f"用户输入: {state['input']}\n是否需要搜索？回答 'yes' 或 'no'。"
response = llm.invoke(prompt).content
state["needs_search"]= response.lower()=="yes"
state["output"]="正在处理..."return state
# 节点：执行搜索defcall_search(state: State)-> State:
result = search.invoke(state["input"])
state["output"]= result
return state
# 节点：直接回答defdirect_answer(state: State)-> State:
state["output"]=f"直接回答: {state['input']}"return state
# 条件函数defroute(state: State)->str:return"call_search"if state["needs_search"]else"direct_answer"# 创建图
workflow = StateGraph(State)
workflow.add_node("call_llm", call_llm)
workflow.add_node("call_search", call_search)
workflow.add_node("direct_answer", direct_answer)
workflow.add_edge(START,"call_llm")
workflow.add_conditional_edges("call_llm", route,{"call_search":"call_search","direct_answer":"direct_answer"})
workflow.add_edge("call_search", END)
workflow.add_edge("direct_answer", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"input":"今天的天气","output":"","needs_search":False})print(result)

复制代码

输出（假设 LLM 认为需要搜索）：

{'input':'今天的天气','output':'搜索结果 for 今天的天气: 找到了一些信息！','needs_search':True}

复制代码

6. 结合 LangChain 和 LangSmith

6.1 与 LangChain 集成

LangGraph 通常与 LangChain 一起使用，LangChain 提供：

LLM 封装

工具支持

Prompt 管理

示例：结合 PromptTemplate 和 LLM：

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
prompt = PromptTemplate.from_template("处理用户输入: {input}")
llm = ChatOpenAI(model="gpt-4o-mini")defllm_node(state: State)-> State:
response = llm.invoke(prompt.format(input=state["input"])).content
state["output"]= response
return state

复制代码

6.2 与 LangSmith 集成

LangSmith 是 LangChain 提供的调试和监控工具，可以可视化 LangGraph 的执行过程。

启用 LangSmith

exportLANGCHAIN_TRACING_V2="true"exportLANGCHAIN_API_KEY="your-langsmith-api-key"

复制代码

运行代码

查看日志

7. 实际应用场景示例

以下是一个实际场景：构建一个客服 Agent，能够回答用户问题或调用搜索工具。
场景描述

代码实现

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 定义工具@tooldefsearch(query:str)->str:"""模拟搜索工具"""returnf"搜索结果: {query} 的信息"# 定义状态classState(TypedDict):input:str
output:str
needs_search:bool# 定义节点defdecide_action(state: State)-> State:
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template("用户问题: {input}\n回答 'yes' 如果需要搜索，'no' 如果可以直接回答。")
response = llm.invoke(prompt.format(input=state["input"])).content
state["needs_search"]= response.lower()=="yes"return state
defsearch_node(state: State)-> State:
result = search.invoke(state["input"])
state["output"]= result
return state
defanswer_node(state: State)-> State:
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template("直接回答用户问题: {input}")
response = llm.invoke(prompt.format(input=state["input"])).content
state["output"]= response
return state
# 条件函数defroute(state: State)->str:return"search_node"if state["needs_search"]else"answer_node"# 创建图
workflow = StateGraph(State)
workflow.add_node("decide_action", decide_action)
workflow.add_node("search_node", search_node)
workflow.add_node("answer_node", answer_node)
workflow.add_edge(START,"decide_action")
workflow.add_conditional_edges("decide_action", route,{"search_node":"search_node","answer_node":"answer_node"})
workflow.add_edge("search_node", END)
workflow.add_edge("answer_node", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"input":"什么是 LangGraph?","output":"","needs_search":False})print(result)

复制代码

输出（假设 LLM 认为需要搜索）：

{'input':'什么是 LangGraph?','output':'搜索结果: 什么是 LangGraph? 的信息','needs_search':True}

复制代码

8. 调试和优化

8.1 使用 LangSmith 调试

8.2 优化技巧

减少 LLM 调用

精简状态

错误处理

9. 常见问题和注意事项

9.1 常见问题

Q：状态未正确更新？

Q：条件边跳转错误？

Q：图执行卡住？

9.2 注意事项

状态不可变

线程安全

工具调用成本

10. 学习资源和下一步

10.1 学习资源

官方文档

LangChain 文档

LangSmith

GitHub 仓库

教程和示例

10.2 下一步

总结

使用 LangGraph 的核心在于：

原文地址：https://blog.csdn.net/u013172930/article/details/147869523

25位大厂高管转战生成式AI创业！吸金猛，不

【LangGraph】如何使用 LangGraph

浏览过的版块

LangGraph——Multi-agent

关于我们

产品与服务

解决方案

产品与服务