在这个极为详细的教程中，我们将探索 **LangGraph** — 一个用于协调复杂多步骤工作流的强大库，适用于大型语言模型（LLMs） — 并将其应用于一个常见的电子商务问题：根据用户的查询决定是否下单或取消订单。到本博客结束时，您将了解如何：
1. 在 Python 环境中设置 LangGraph。 2. 加载和管理数据（例如，库存和客户）。 3. 定义节点（工作流中的单个任务）。 4. 构建节点和边的图，包括条件分支。 5. 可视化和测试工作流。
我们将逐步进行，详细解释每个概念 — 非常适合初学者和希望使用 LLMs 构建动态或循环工作流的人，我还提供了数据集的链接供您尝试。
目录
什么是 LangGraph?
问题陈述：订单管理
导入说明
数据加载和状态定义
创建工具和 LLM 集成
定义工作流节点
构建工作流图
可视化和测试工作流
结论
什么是 LangGraph？
LangGraph 是一个库，它为 LangChain 工作流带来了基于图的方式。传统的管道通常线性地从一个步骤移动到另一个步骤，但现实世界的任务常常需要分支、条件逻辑，甚至循环（重试失败的步骤、澄清用户输入等）。
LangGraph 的主要特点：
• 节点：单个任务或功能（例如，检查库存、计算运费）。 • 边：定义节点之间数据和控制的流动。可以是条件性的。 • 共享状态：每个节点可以返回更新全局状态对象的数据，避免手动传递数据。 • 工具集成：轻松整合外部工具或函数，供 LLM 调用。 • 人机协作（可选）：插入需要人工审核的节点。
问题陈述：订单管理
在此场景中，用户的查询可以是关于下新订单或取消现有订单：
• PlaceOrder：检查商品可用性，计算运费，并模拟支付。 • CancelOrder：提取 order_id并将订单标记为已取消。
因为我们需要分支（决定“PlaceOrder”与“CancelOrder”），我们将使用 LangGraph 创建一个条件流程：
1. 分类 查询。 2. 如果 PlaceOrder，则移动到检查库存、运费和支付。 3. 如果 CancelOrder，则解析出 order_id并调用取消工具。
导入说明
下面是您提供的代码的确切第一部分，展示了导入和环境设置。我们在代码后添加了注释以解释每个部分。
### Import required libraries
import os
import pandas as pd
import random
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.runnables.graph import MermaidDrawMethod
from IPython.display import display, Image
from typing importLiteral
from langchain_core.prompts import ChatPromptTemplate
from typing importDict, TypedDict
### Load environment variables
os.environ["OPENAI_API_KEY"] = ""langchain_core.tools, langchain_openai, ToolNode, 等：
• tool（一个装饰器）将 Python 函数转换为 LLM 可以调用的“工具”。• ChatOpenAI是我们与 GPT 模型进行交互的 LLM 客户端。• ToolNode是来自langgraph.prebuilt的一个预构建节点，负责工具执行。• StateGraph、MessagesState、START、END来自langgraph.graph——它们对于定义我们的工作流程至关重要。• MermaidDrawMethod帮助将工作流程可视化为 Mermaid.js 图表。
数据加载和状态定义
数据链接：Data
在下一个代码片段中，我们加载 CSV 文件（用于库存和客户）并转换它们为字典。我们还定义了我们的状态类型字典。
### Load datasets
inventory_df = pd.read_csv("inventory.csv")
customers_df = pd.read_csv("customers.csv")
### Convert datasets to dictionaries
inventory = inventory_df.set_index("item_id").T.to_dict()
customers = customers_df.set_index("customer_id").T.to_dict()
classState(TypedDict):
query: str
category: str
next_node: str
item_id: str
order_status: str
cost: str
payment_status: str
location: str
quantity: intCSV 到字典：
• inventory和customers是以item_id或customer_id为键的字典。这使得查找像inventory[item_51]这样的操作变得简单。
状态：
• 一个类型字典，以便我们知道期望哪些字段。例如， query、category、item_id等。• category通常是“PlaceOrder”或“CancelOrder”。• next_node可以存储下一个节点名称，尽管我们依赖图的边缘进行转换。• 这有助于在一个对象中跟踪一切——库存检查、支付状态 等。
创建工具和LLM集成
现在我们定义我们的LLM和工具。这里的主要工具是cancel_order，它使用LLM从查询中提取order_id。
@tool
defcancel_order(query: str) -> dict:
"""Simulate order cancelling"""
order_id = llm.with_structured_output(method='json_mode').invoke(f'Extract order_id from the following text in json format: {query}')['order_id']
#amount = query.get("amount")
ifnot order_id:
return {"error": "Missing 'order_id'."}
return {"order_status": "Order stands cancelled"}
### Initialize LLM and bind tools
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools_2 = [cancel_order]
llm_with_tools_2 = llm.bind_tools(tools_2)
tool_node_2 = ToolNode(tools_2)@tool:
• cancel_order函数现在是LLM可以调用的工具，如果它决定需要取消订单的话。
提取order_id：
• 我们调用 llm.with_structured_output(method='json_mode')来指示LLM返回JSON。然后我们解析出'order_id'。
LLM初始化：
• model="gpt-4o-mini"是选择的模型，temperature=0用于确定性响应。
绑定和ToolNode：
• llm.bind_tools(tools_2)将我们的LLM与cancel_order工具连接起来。• ToolNode是一个专门的节点，可以自动处理这些绑定的工具。
定义工作流节点
我们现在将开始逐个定义节点。
模型调用节点
这些节点可以调用模型
def call_model_2(state: MessagesState):
"""Use the LLM to decide the next step."""
messages = state["messages"]
response = llm_with_tools_2.invoke(str(messages))
return {"messages": [response]}
defcall_tools_2(state: MessagesState) -> Literal["tools_2", END]:
"""Route workflow based on tool calls."""
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return"tools_2"
return END• call_model_2: 接收对话（messages）并将其传递给带工具的LLM。如果LLM触发了工具调用，我们将在call_tools_2中检测到。• call_tools_2: 检查LLM是否请求了工具调用（tool_calls）。如果是，我们将路由到"tools_2"，即ToolNode；否则，我们结束工作流。
分类查询
在这里，我们定义一个节点来对查询进行分类：
def categorize_query(state: MessagesState) -> MessagesState:
"""Categorize user query into PlaceOrder or CancelOrder"""
prompt = ChatPromptTemplate.from_template(
"Categorize user query into PlaceOrder or CancelOrder"
"Respond with either 'PlaceOrder', 'CancelOrder' Query: {state}"
)
chain = prompt | ChatOpenAI(temperature=0)
category = chain.invoke({"state": state}).content
return {"query":state,"category": category}• 该节点使用LLM对用户的查询进行分类。返回值在状态中设置了 "category"。
检查库存
def check_inventory(state: MessagesState) -> MessagesState:
"""Check if the requested item is in stock."""
item_id = llm.with_structured_output(method='json_mode').invoke(f'Extract item_id from the following text in json format: {state}')['item_id']
quantity = llm.with_structured_output(method='json_mode').invoke(f'Extract quantity from the following text in json format: {state}')['quantity']
if not item_id or not quantity:
return {"error": "Missing 'item_id' or 'quantity'."}
if inventory.get(item_id, {}).get("stock", 0) >= quantity:
print("IN STOCK")
return {"status": "In Stock"}
return {"query":state,"order_status": "Out of Stock"}• 尝试从对话中解析 item_id和quantity。• 检查 inventory[item_id]["stock"]以确认可用性。
计算运费
我们为特定客户定义了一个计算运费的节点
def compute_shipping(state: MessagesState) -> MessagesState:
"""Calculate shipping costs."""
item_id = llm.with_structured_output(method='json_mode').invoke(f'Extract item_id from the following text in json format: {state}')['item_id']
quantity = llm.with_structured_output(method='json_mode').invoke(f'Extract quantity from the following text in json format: {state}')['quantity']
customer_id = llm.with_structured_output(method='json_mode').invoke(f'Extract customer_id from the following text in json format: {state}')['customer_id']
location = customers[customer_id]['location']
ifnot item_id ornot quantity ornot location:
return {"error": "Missing 'item_id', 'quantity', or 'location'."}
weight_per_item = inventory[item_id]["weight"]
total_weight = weight_per_item * quantity
rates = {"local": 5, "domestic": 10, "international": 20}
cost = total_weight * rates.get(location, 10)
print(cost,location)
return {"query":state,"cost": f"${cost:.2f}"}• 从用户的查询中检索 customer_id，然后在 customers字典中查找他们的 location。• 根据商品的重量、数量和用户的地点计算运费。
处理支付
我们将定义一个用于处理支付的节点：
def process_payment(state: State) -> State:
"""Simulate payment processing."""
cost = llm.with_structured_output(method='json_mode').invoke(f'Extract cost from the following text in json format: {state}')
if not cost:
return {"error": "Missing 'amount'."}
print(f"PAYMENT PROCESSED: {cost} and order successfully placed!")
payment_outcome = random.choice(["Success", "Failed"])
return {"payment_status": payment_outcome}• 使用 random.choice来模拟成功或失败。• 在生产系统中，您需要与真实的支付网关集成。
路由功能
我们现在定义一个用于路由查询的节点：
def route_query_1(state: State) -> str:
"""Route the query based on its category."""
print(state)
if state["category"] == "PlaceOrder":
return "PlaceOrder"
elif state["category"] == "CancelOrder":
return "CancelOrder"• 决定接下来要遵循的路径：“PlaceOrder”或“CancelOrder”。在LangGraph中，我们将“PlaceOrder”映射到CheckInventory节点，将“CancelOrder”映射到CancelOrder节点。
构建工作流图
下面，我们创建一个 StateGraph，添加节点，并定义边和条件边。
### Create the workflow
workflow = StateGraph(MessagesState)
#Add nodes
workflow.add_node("RouteQuery", categorize_query)
workflow.add_node("CheckInventory", check_inventory)
workflow.add_node("ComputeShipping", compute_shipping)
workflow.add_node("ProcessPayment", process_payment)
workflow.add_conditional_edges(
"RouteQuery",
route_query_1,
{
"PlaceOrder": "CheckInventory",
"CancelOrder": "CancelOrder"
}
)
workflow.add_node("CancelOrder", call_model_2)
workflow.add_node("tools_2", tool_node_2)
### Define edges
workflow.add_edge(START, "RouteQuery")
workflow.add_edge("CheckInventory", "ComputeShipping")
workflow.add_edge("ComputeShipping", "ProcessPayment")
workflow.add_conditional_edges("CancelOrder", call_tools_2)
workflow.add_edge("tools_2", "CancelOrder")
workflow.add_edge("ProcessPayment", END)StateGraph(MessagesState):
• 我们指定 MessagesState来保存对话数据。
节点：
• RouteQuery是分类用户意图的入口节点。• “CheckInventory”、“ComputeShipping”和“ProcessPayment”处理下单流程。 • “CancelOrder”和“tools_2”处理取消订单流程。
条件边：
• 调用 workflow.add_conditional_edges("RouteQuery", route_query_1, ...)确保我们在“下单”的情况下转到 CheckInventory，或者在“取消订单”的情况下转到 CancelOrder。
循环：
• 当用户点击“取消订单”时，我们检查 LLM 是否触发了工具调用 ( call_tools_2)。如果是，我们转到tools_2（ToolNode）；在工具被调用后，它会返回到“取消订单”，给 LLM 机会产生进一步的动作或结束。
结束：
• “ProcessPayment” 导致 END，结束“下单”路径。
可视化和测试工作流
下一个代码片段将工作流编译成一个代理，渲染为Mermaid图，并使用示例查询进行测试。
### Compile the workflow
agent = workflow.compile()
### Visualize the workflow
mermaid_graph = agent.get_graph()
mermaid_png = mermaid_graph.draw_mermaid_png(draw_method=MermaidDrawMethod.API)
display(Image(mermaid_png))
### Query the workflow
user_query = "I wish to cancel order_id 223"
for chunk in agent.stream(
{"messages": [("user", user_query)]},
stream_mode="values",
):
chunk["messages"][-1].pretty_print()
auser_query = "customer_id: customer_14 : I wish to place order for item_51 with order quantity as 4 and domestic"
for chunk in agent.stream(
{"messages": [("user", auser_query)]},
stream_mode="values",
):
chunk["messages"][-1].pretty_print()编译：
• agent = workflow.compile()将我们的节点/边定义转换为可执行的代理。
可视化：
• 我们得到一个Mermaid图（ mermaid_png），可以在Jupyter笔记本中显示，以便进行调试或演示。
测试查询：
• 第一个测试：“I wish to cancel order_id 223” 应该路由到 CancelOrder。
• 第二个测试：“customer_id: customer_14 : I wish to place order for item_51…” 应该路由到下单工作流。
结论：
通过利用 LangGraph，我们构建了一个动态的分支工作流，该工作流根据用户意图来下单或取消订单。我们展示了：
• 如何使用 LLM 节点 ( categorize_query) 对查询进行分类。• 如何绑定工具 ( cancel_order) 并将其集成到工作流中。• 如何通过独立节点检查库存、计算运费和处理付款。 • 如何使用 Mermaid.js 可视化整个工作流。
这种方法是可扩展的：您可以添加更多步骤（例如，地址验证、促销代码）或额外的分支（例如，更新现有订单），而无需重写单体脚本。如果您需要 循环 来重试失败的付款或验证用户确认，LangGraph 也可以处理。
