：Langgraph应用，执行流程由线转图)
简介
Langgraph是langchain框架提供的一个组件，langgraph能够解决AI执行流程中迭代、循环或者根据结果返回上一步，与之前讲的chain链相比，能够实现更加复杂的AI执行流。
langgraph
从chain转到langgraph从数学的角度上来讲，执行流从线性流程转到了流程图。
langgraph的组成主要有三部分：
Langgraph=节点+边+状态
节点：一个节点就是一个执行单元，相当于一次函数的调用。（可以是一次模型的调用，一次搜索，一次加密等等）
边：边能够连接一个个节点，它决定了下一个应该去到哪个节点执行
状态：实现数据共享，是实现AI短期记忆的灵魂
乍一看好像有些云里雾里的，我打个比方：玩家（状态），在玩一个大富翁，每个节点和边组成地图，玩家初始资金（数据）有1000块钱，玩家每走一格可能会发生一些事件，比如说后退一步，被小偷偷300块钱，买房子等等，这些事件相当于节点，走的方向相当于边，最后玩家成功走到了终点END，你可以得知玩家（状态）最后还有多少钱，有多少资产。
下面用一个示例来演示。
改造RAG知识库
示例
这个是之前文章中构建知识库的相关代码，本次用langgraph改造流程
/* by 01130.hk - online tools website : 01130.hk/zh/htaccess2nginx.html */ 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 ToolNode os.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())) #创建state class TaskState(TypedDict): messages: Annotated[list[BaseMessage], operator.add] #创建node def 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的构建，对于其他的细节，请看前面文章。
代码流程如下：
初始化工具集->定义状态，定义条件边，节点->构建节点->连接边->构建图->运行图
初始化工具集
这个前面文章有，就不废话了。
定义状态，定义条件边，节点
状态
/* by 01130.hk - online tools website : 01130.hk/zh/htaccess2nginx.html */ #创建state class 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)运行跟之前运行普通模型一样
stream方法会返回每个节点中的状态（上面定义的类）invoke方法直接返回最终状态
langgraph是实现多Agent协作的核心，下一篇文章会讲如何多agent协作
如果❤喜欢❤本系列教程，就点个关注吧，后续不定期更新~
