AI创想

标题: 【LangGraph】如何使用 LangGraph [打印本页]

---

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

---

目录

安装和环境准备LangGraph 的核心概念回顾构建一个 LangGraph 工作流：分步指南代码示例：简单工作流高级功能：条件边、循环和工具调用结合 LangChain 和 LangSmith实际应用场景示例调试和优化常见问题和注意事项学习资源和下一步

---

1. 安装和环境准备

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

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

1. pip install langchain langgraph

复制代码

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

1. pip install langchain-openai  # 用于 OpenAI API

复制代码

1.2 配置 API 密钥

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

1. exportOPENAI_API_KEY="your-api-key"

复制代码

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

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

复制代码

1.3 验证安装

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

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

复制代码

---

2. LangGraph 的核心概念回顾

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

状态（State）：一个数据结构（如字典或类），用于存储工作流的上下文信息，在节点间传递和更新。节点（Node）：执行具体任务的函数（如调用 LLM、处理数据），接收状态并返回更新后的状态。边（Edge）：定义节点之间的连接，可以是无条件的（直接跳转）或条件边（基于状态跳转）。图（Graph）：由节点和边组成的结构，表示整个工作流，使用 StateGraph 类定义。

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

---

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

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

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

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

复制代码

3.2 创建节点

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

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

复制代码

3.3 定义边

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

add_edge 添加无条件边。add_conditional_edges 添加条件边。

3.4 构建图

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

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

复制代码

3.5 执行工作流

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

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

复制代码

---

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

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

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

复制代码

输出：

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

复制代码

这个示例展示了：

一个简单的状态（包含 input 和 output）。两个节点：process_input 和 finalize_output。线性工作流：从 START 到 process_input，再到 finalize_output，最后到 END。

---

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

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

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

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

复制代码

输出（输入 “Hi”）：

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

复制代码

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

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

复制代码

5.2 循环

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

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

复制代码

输出：

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

复制代码

5.3 工具调用

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

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

复制代码

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

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

复制代码

---

6. 结合 LangChain 和 LangSmith

6.1 与 LangChain 集成

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

LLM 封装：如 ChatOpenAI、ChatAnthropic。工具支持：通过 langchain_core.tools 定义工具。Prompt 管理：使用 PromptTemplate 构建结构化提示。

示例：结合 PromptTemplate 和 LLM：

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

复制代码

6.2 与 LangSmith 集成

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

启用 LangSmith：
1. exportLANGCHAIN_TRACING_V2="true"exportLANGCHAIN_API_KEY="your-langsmith-api-key"
复制代码
运行代码：LangGraph 会自动将执行日志发送到 LangSmith。
查看日志：在 LangSmith 界面中，可以看到每个节点的输入输出、状态变化和执行路径。

---

7. 实际应用场景示例

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

用户输入问题。Agent 判断是否需要搜索。如果需要搜索，调用搜索工具并生成回答。如果无需搜索，直接生成回答。

代码实现

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

复制代码

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

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

复制代码

---

8. 调试和优化

8.1 使用 LangSmith 调试

检查每个节点的输入输出。分析条件边的跳转逻辑。识别性能瓶颈（如 LLM 调用时间过长）。

8.2 优化技巧

减少 LLM 调用：尽量在节点中缓存结果或合并逻辑。精简状态：只存储必要的数据，减少内存占用。错误处理：在节点中添加 try-except，处理工具调用或 LLM 失败的情况。

---

9. 常见问题和注意事项

9.1 常见问题

Q：状态未正确更新？

A：确保节点返回了更新后的状态，且状态结构与定义一致。

Q：条件边跳转错误？

A：检查条件函数的返回值是否与 add_conditional_edges 的映射一致。

Q：图执行卡住？

A：可能是循环未正确终止，检查条件边逻辑。

9.2 注意事项

状态不可变：LangGraph 默认状态是可变的，建议在节点中创建状态副本以避免意外修改。线程安全：如果在多线程环境中使用，确保状态和节点函数是线程安全的。工具调用成本：频繁调用外部工具（如搜索 API）可能增加延迟和费用，需优化调用频率。

---

10. 学习资源和下一步

10.1 学习资源

官方文档：https://langchain-ai.github.io/langgraph/LangChain 文档：https://python.langchain.com/docs/LangSmith：https://smith.langchain.com/GitHub 仓库：https://github.com/langchain-ai/langgraph教程和示例：LangChain 官方博客和 YouTube 频道。

10.2 下一步

尝试更复杂的场景，如多 Agent 协作或 RAG 工作流。探索 LangGraph 的异步支持（graph.ainvoke），提高性能。使用 LangSmith 优化你的工作流，分析性能瓶颈。

---

总结

使用 LangGraph 的核心在于：

定义清晰的状态结构。编写模块化的节点函数。通过边和条件边实现动态逻辑。结合 LangChain 的 LLM 和工具，构建强大的工作流。

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

---

欢迎光临 AI创想 (https://www.llms-ai.com/)

Powered by Discuz! X3.4