当前位置：首页 > news >正文

langchain 快速入门(五)：Langgraph应用，执行流程由线转图

news 2026/3/27 4:12:01

简介

Langgraph是langchain框架提供的一个组件，langgraph能够解决AI执行流程中迭代、循环或者根据结果返回上一步，与之前讲的chain链相比，能够实现更加复杂的AI执行流。

langgraph

从chain转到langgraph从数学的角度上来讲，执行流从线性流程转到了流程图。

langgraph的组成主要有三部分：
Langgraph=节点+边+状态
节点： 一个节点就是一个执行单元，相当于一次函数的调用。（可以是一次模型的调用，一次搜索，一次加密等等）
边：边能够连接一个个节点，它决定了下一个应该去到哪个节点执行
状态： 实现数据共享，是实现AI短期记忆的灵魂

乍一看好像有些云里雾里的，我打个比方：玩家（状态），在玩一个大富翁，每个节点和边组成地图，玩家初始资金（数据）有1000块钱，玩家每走一格可能会发生一些事件，比如说后退一步，被小偷偷300块钱，买房子等等，这些事件相当于节点，走的方向相当于边，最后玩家成功走到了终点END，你可以得知玩家（状态）最后还有多少钱，有多少资产。

下面用一个示例来演示。

改造RAG知识库

示例

这个是之前文章中构建知识库的相关代码，本次用langgraph改造流程

import os
import operator
from typing import TypedDict, Annotated
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage, BaseMessage
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.graph import StateGraph, END
from langgraph.prebuilt.tool_node import ToolNodeos.environ["DASHSCOPE_API_KEY"] = "sk-b07b9caab2144ff096002f81cb70b0af"@tool
def calculator(expression: str) -> str:"""计算数学表达式。需要精确计算时使用。参数:expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。返回:str: 计算结果，如 "4.0" 或 "400.0"。"""print(f"[🛠️ 工具调用] 计算器正在计算: {expression}")try:return str(eval(expression))except Exception as e:return f"计算错误: {e}"@tool
def rag_search(query: str) -> str:"""从数据库中搜索与查询公司内部相关的文档，包括公司计划名，代号，截止日期等详细信息。参数:query (str): 要搜索的查询字符串。返回:str: 与查询相关的文档内容。"""# 从RAG数据库中检索文档raw_text = """【公司内部机密：代号“深蓝计划”】1. 项目目标：开发一款能听懂猫语的翻译器。2. 核心技术：基于Transformer的“喵声波”分析算法。3. 团队暗号：如果有人问“今天天气怎么样？”，必须回答“我想吃鱼”。4. 截止日期：2026年12月31日。5. 经费预算：仅剩50元人民币，主要用于购买猫条。"""RAG_PATH = "faiss_index"docs = [Document(page_content=raw_text)]text_splitter = RecursiveCharacterTextSplitter(chunk_size=25, chunk_overlap=5)split_docs = text_splitter.split_documents(docs)embeddings = DashScopeEmbeddings(model="text-embedding-v1")if os.path.exists(RAG_PATH):print("公司内部数据库已存在")ragdb = FAISS.load_local(RAG_PATH, embeddings, allow_dangerous_deserialization=True)else:print("创建公司内部数据库")ragdb = FAISS.from_documents(split_docs, embeddings)ragdb.save_local(RAG_PATH)return "\n\n".join(doc.page_content for doc in ragdb.similarity_search(query, k=2))#构造agent流程图def Init_Agent():#初始化模型tool_maps={"rag_search": rag_search,"calculator": calculator}llm = ChatTongyi(model_name="qwen-plus")tool_llm = llm.bind_tools(tools=list(tool_maps.values()))#创建stateclass TaskState(TypedDict):messages: Annotated[list[BaseMessage], operator.add]#创建nodedef agent_node(state: TaskState):"""节点：思考 (Think)接收当前状态，调用 LLM，返回新消息"""messages = state["messages"]response = tool_llm.invoke(messages)return {"messages": [response]}#定义边def condition_tools(state: TaskState):"""节点：工具 (Tool)接收当前状态，调用工具，返回新消息"""messages = state["messages"][-1]if messages.tool_calls:return "tool_node"else:return END#添加边workflow = StateGraph(TaskState)workflow.add_node("agent_node", agent_node)workflow.add_node("tool_node", ToolNode(tool_maps.values()))workflow.add_conditional_edges("agent_node", condition_tools, {"tool_node": "tool_node",END: END})workflow.add_edge("tool_node", "agent_node")workflow.set_entry_point("agent_node")return workflow.compile()if __name__ == "__main__":app = Init_Agent()input = "公司的经费预算是多少，如果预算预算提高46%后多少"for event in app.stream({"messages": [HumanMessage(content=input)]}):for key, value in event.items():print(f"\n[{key}]")print(value["messages"][-1].content)

代码解释

本次代码中重点讲langgraph的构建，对于其他的细节，请看前面文章。
代码流程如下：
初始化工具集->定义状态，定义条件边，节点->构建节点->连接边->构建图->运行图

初始化工具集

这个前面文章有，就不废话了。

定义状态，定义条件边，节点

状态

#创建stateclass TaskState(TypedDict):messages: Annotated[list[BaseMessage], operator.add]

状态是TypedDict的子类（字典）。
上面的BaseMessage是ToolMessage,AIMessage,HumanMessage等的父类，这个list主要用于存放每个节点的历史消息（短期记忆）
Annotated[..., operator.add]表示追加，将节点返回的消息追加到后面，而不是覆盖。
格式如下（可以创建多个自定义字段）：

class  StateName(TypedDict):fieldName: fieldType

条件边

def condition_tools(state: TaskState):"""节点：工具 (Tool)接收当前状态，调用工具，返回新消息"""messages = state["messages"][-1]if messages.tool_calls:return "tool_node"else:return END

返回值END和"tool_node"表示定义的节点名称，END默认是结束节点
格式如下：

def EdgeName(state: StateClass)return "NextNode"

节点

@tool
def calculator(expression: str) -> str:......@tool
def rag_search(query: str) -> str:......def agent_node(state: TaskState):......

节点可以是工具函数，也可以是普通函数（普通函数需要用state传入）

构建节点

workflow = StateGraph(TaskState)
workflow.add_node("agent_node", agent_node)
workflow.add_node("tool_node", ToolNode(tool_maps.values()))

StateGraph(TaskState)初始化图，将刚刚创建的状态传入
add_node方法是创建节点"tool_node"节点名称（自定义用于标识节点），agent_node创建的节点函数
ToolNode是langchain提供的创建工具节点的函数，帮我们完成了调用工具集，更新状态的全过程（不用这个需要我们自己手动创建工具循环节点，比较麻烦，参考之前文章）

连接边

workflow.add_conditional_edges("agent_node", condition_tools, {"tool_node": "tool_node",END: END})workflow.add_edge("tool_node", "agent_node")

add_conditional_edges创建条件边方法（分支），根据返回内容决定节点走向
add_edge固定走向，如上tool_node->agent_node

构建图

workflow.set_entry_point("agent_node")
workflow.compile()

set_entry_point确定图的入口
compile构建图

运行图

if __name__ == "__main__":app = Init_Agent()input = "公司的经费预算是多少，如果预算预算提高46%后多少"for event in app.stream({"messages": [HumanMessage(content=input)]}):for key, value in event.items():print(f"\n[{key}]")print(value["messages"][-1].content)

运行跟之前运行普通模型一样