FEATURED · 精选文章

LangChain与LangGraph技术解析:从框架到智能体工作流编排

发布时间 / 2026/8/6 7:40:49
来源 / 创域科博编辑部
栏目 / 资讯中心
LangChain与LangGraph技术解析:从框架到智能体工作流编排 这次我们来看一个关于 LangChain 和 LangGraph 的技术解析。如果你正在接触大模型应用开发对这两个名字感到困惑不清楚它们到底是并列、替代还是从属关系以及在实际项目中该如何选择那么这篇文章就是为你准备的。本文不会堆砌复杂概念而是直接切入核心LangChain 和 LangGraph 分别是什么它们解决什么问题在什么场景下该用谁以及如何快速上手验证。我们将通过一个简单的“动画讲解”类比帮你理清思路并给出可操作的技术路径。简单来说LangChain 是一个用于构建基于大语言模型LLM应用程序的框架它提供了丰富的组件和工具链帮你连接模型、数据源和外部工具。而 LangGraph 是 LangChain 生态系统中的一个库它专注于一件事构建有状态的、多步骤的、复杂的工作流特别是那些带有循环、分支和状态管理的智能体Agent应用。你可以把 LangChain 想象成一个功能齐全的“工具箱”而 LangGraph 是这个工具箱里专门用于设计和组装复杂“自动化流水线”的精密“图纸和连接器”。对于开发者而言最需要关注的是如果你的应用只是简单的问答、文档检索或单次模型调用LangChain 的基础链Chain就足够了。但如果你需要构建一个能自主决策、根据中间结果循环执行不同工具、并保持对话状态的智能体那么 LangGraph 就是你不可或缺的利器。它让复杂工作流的编排变得像画流程图一样直观。1. 核心能力速览为了让你快速把握核心我们将两者的关键信息对比如下能力项LangChainLangGraph项目定位大模型应用开发框架基于图的智能体/工作流编排库核心功能模型集成、提示模板、记忆、检索、工具调用、链Chain有状态、多步骤工作流编排支持循环、条件分支与对方关系父框架/生态系统LangChain 生态系统内的一个核心库关键抽象Chain链 按固定顺序执行Graph图 State状态 动态执行路径适用场景检索增强生成RAG、聊天机器人、文档分析、简单任务自动化复杂决策智能体、多工具协作流程、需长期记忆的对话系统、业务流程自动化上手门槛中等需理解其模块化概念中高需理解状态管理和图节点概念启动/使用方式通过pip install langchain安装在Python代码中导入使用通过pip install langgraph安装通常与langchain协同使用“硬件”门槛无特殊要求依赖主要是网络调用API或本地算力运行本地模型同LangChain无额外硬件要求复杂性体现在软件设计上“显存/资源”占用取决于集成的LLM如调用OpenAI API则无本地显存占用运行本地模型则依赖模型大小工作流本身资源消耗极低资源占用主体仍是被调用的LLM和工具是否支持“批量任务”可通过自定义链或利用框架组件实现其图结构非常适合描述和运行批量、并行的复杂任务流是否支持“接口API”可轻松封装成FastAPI等Web服务其工作流可作为后端核心逻辑通过API暴露2. 适用场景与使用边界理解一个技术的最佳方式就是看它用在哪儿。下面我们来具体拆解它们的适用场景。2.1 何时选择 LangChainLangChain 是一个“面”上的框架适合构建绝大多数标准的 LLM 应用。当你面临以下场景时应优先考虑 LangChain检索增强生成RAG这是 LangChain 的招牌场景。你需要从自己的文档PDF、Word、数据库中检索信息然后交给 LLM 生成答案。LangChain 提供了完整的文档加载、切分、向量化、检索和生成链。基础聊天机器人需要连接聊天历史记忆、处理用户消息、调用 LLM 并返回回复。LangChain 的ConversationChain、LCELLangChain Expression Language可以优雅地处理。工具增强的简单助手让 LLM 能够使用计算器、搜索网络、查询数据库等。LangChain 的Tool和Agent基础架构可以快速实现。标准化任务流水线例如一个固定的处理流程加载文档-提取关键信息-格式化-保存。你可以用SequentialChain来定义这个顺序。LangChain 的边界当你的应用逻辑需要大量的“如果...那么...”判断或者需要根据 LLM 的中间输出反复执行某些步骤时用基础的Chain会变得非常笨拙和难以维护。这时你就碰到了 LangChain 基础能力的边界。2.2 何时选择 LangGraphLangGraph 是一个“点”上的深化工具专攻复杂、有状态的工作流。当你的需求具备以下特征时LangGraph 是更优解构建复杂智能体Agent智能体需要自主规划、执行工具、评估结果、并决定下一步是继续、重试还是结束。这个过程天然带有循环和分支。LangGraph 的StateGraph是为此而生。需要持久化且结构化的状态比如一个订票智能体状态里需要维护用户查询的航班信息、选择的座位、支付状态等。LangGraph 的State对象可以清晰地定义和管理这些状态。工作流包含明确的人工或系统审批节点例如一个内容生成流程生成草稿-提交审核- 判断审核通过- 是则发布否则返回修改。这种带条件分支的流程用图来描绘非常直观。模拟多角色协作你可以定义不同的“工位”节点如“分析师”、“决策者”、“执行者”让状态和数据在它们之间流转模拟一个团队协作的过程。LangGraph 的边界对于简单的、一次性的、无状态的任务使用 LangGraph 属于“杀鸡用牛刀”会引入不必要的复杂性。它更适合作为复杂系统的核心编排引擎。重要合规提醒无论是使用 LangChain 还是 LangGraph当集成 LLM 处理用户数据、生成内容或执行自动化操作时必须注意数据隐私确保用户数据的收集、处理符合相关法律法规敏感信息需脱敏。内容安全对 LLM 生成的内容进行必要的审核和过滤避免产生有害、偏见或侵权内容。工具调用安全对智能体可调用的工具如数据库写入、发送邮件做严格的权限控制和操作确认防止未经授权的操作。3. 环境准备与前置条件开始动手之前你需要准备好基础环境。由于 LangChain 和 LangGraph 是 Python 库因此核心要求是 Python 环境。Python 版本推荐使用 Python 3.8 及以上版本。可以使用python --version检查。包管理工具使用pip进行安装。建议使用虚拟环境如venv或conda来隔离项目依赖。# 创建虚拟环境可选但推荐 python -m venv langchain-env # 激活虚拟环境 # Windows: langchain-env\Scripts\activate # macOS/Linux: source langchain-env/bin/activateLLM 接入准备你需要一个 LLM 的调用权限。这通常是云 API如 OpenAI GPT、 Anthropic Claude、 智谱AI、 月之暗面等。你需要准备相应的 API Key。本地模型如通过 Ollama、 vLLM、 Transformers 库部署的本地大模型。你需要确保有足够的硬件资源GPU 显存/系统内存。网络访问如果使用云 API需要保证网络能够稳定访问对应服务。4. 安装部署与启动方式安装过程非常简单因为它们就是标准的 Python 包。4.1 安装 LangChain打开终端在激活的虚拟环境中运行pip install langchain这将安装 LangChain 的核心库。根据你需要的功能可能还要安装额外的集成包例如用于 OpenAI 的pip install openai或者用于向量数据库的pip install chromadb4.2 安装 LangGraphLangGraph 需要单独安装pip install langgraph请注意langgraph通常与langchain一起使用因为它依赖于 LangChain 的一些核心概念如Runnable、Messages。4.3 验证安装创建一个简单的 Python 脚本test_install.py来验证import langchain import langgraph print(fLangChain version: {langchain.__version__}) print(fLangGraph version: {langgraph.__version__}) print(安装成功)运行python test_install.py如果没有报错并输出版本号说明环境准备就绪。5. 功能测试与效果验证让我们通过两个具体的例子来感受 LangChain 的“链”和 LangGraph 的“图”在用法和思维上的根本不同。5.1 LangChain 基础链测试一个简单的串联任务我们将创建一个链它先让 LLM 生成一个笑话然后让 LLM 评价这个笑话好不好笑。# test_langchain_chain.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 1. 初始化模型 (请替换你的API Key) model ChatOpenAI(modelgpt-3.5-turbo, api_keyyour-api-key-here) # 2. 定义第一个提示词生成笑话 joke_prompt ChatPromptTemplate.from_template(请讲一个关于{topic}的笑话。) # 3. 定义第二个提示词评价笑话 critique_prompt ChatPromptTemplate.from_template( 请评价以下笑话是否好笑并简要说明原因。 笑话{joke} 评价 ) # 4. 使用 LCEL 将组件连接成链 chain ( {topic: RunnablePassthrough()} # 接收初始输入 topic | joke_prompt # 传递给笑话提示词 | model # 调用模型生成笑话 | StrOutputParser() # 解析模型输出为字符串 | {joke: RunnablePassthrough()} # 将笑话作为字典的 joke 键值 | critique_prompt # 将包含 joke 的字典传递给评价提示词 | model # 再次调用模型进行评价 | StrOutputParser() # 解析最终评价 ) # 5. 运行链 result chain.invoke(程序员) print(生成的笑话及评价) print(result)预期效果这段代码会输出一个关于“程序员”的笑话以及模型对这个笑话的评价。这是一个典型的顺序执行链数据流是线性的、预先定义好的。5.2 LangGraph 工作流测试一个带循环的智能体我们将模拟一个更复杂的场景一个“研究助手”智能体。它根据用户问题决定是调用“网络搜索”工具还是直接回答。如果搜索了它还可以决定是否需要基于搜索结果进行更深入的二次搜索。# test_langgraph_agent.py from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 对话消息历史 question: str # 用户原始问题 search_results: list[str] # 搜索工具返回的结果 decision: Literal[answer, search, more_search] # 智能体的决策 # 2. 初始化模型 model ChatOpenAI(modelgpt-3.5-turbo, api_keyyour-api-key-here) # 3. 定义节点函数 def call_llm(state: AgentState): 调用LLM分析当前状态并做出决策 system_msg SystemMessage(content你是一个研究助手。根据已有信息决定下一步是直接回答(answer)进行搜索(search)还是基于现有结果深入搜索(more_search)。) history state[messages] prompt f 用户问题{state[question]} 已有搜索结果{state.get(search_results, [])} 请只输出你的决策必须是answer, search, more_search 中的一个。 human_msg HumanMessage(contentprompt) response model.invoke([system_msg, human_msg]) decision response.content.strip().lower() return {decision: decision} def search_tool(state: AgentState): 模拟网络搜索工具此处用固定文本模拟 print(f[模拟搜索] 搜索关键词: {state[question]}) # 模拟返回搜索结果 simulated_results [ f关于{state[question]}的初步资料1。, f关于{state[question]}的初步资料2。 ] return {search_results: simulated_results} def answer_tool(state: AgentState): 基于已有信息生成最终答案 context .join(state.get(search_results, [])) prompt f请根据以下信息回答用户问题。\n信息{context}\n问题{state[question]}\n答案 response model.invoke([HumanMessage(contentprompt)]) return {messages: [HumanMessage(contentf答案{response.content})]} # 4. 构建图 workflow StateGraph(AgentState) # 5. 添加节点 workflow.add_node(analyze, call_llm) # 分析决策节点 workflow.add_node(search, search_tool) # 搜索节点 workflow.add_node(answer, answer_tool) # 回答节点 # 6. 设置入口点 workflow.set_entry_point(analyze) # 7. 定义边条件路由 def decide_route(state: AgentState): 根据决策结果路由到下一个节点 decision state[decision] if decision answer: return answer elif decision search: return search elif decision more_search: return search # 再次进入搜索节点 else: return END # 未知决策结束 workflow.add_conditional_edges( analyze, # 源节点 decide_route, # 路由函数 { answer: answer, search: search, more_search: search, } ) workflow.add_edge(search, analyze) # 搜索完后回到分析节点重新决策 workflow.add_edge(answer, END) # 给出答案后结束 # 8. 编译图 app workflow.compile() # 9. 运行图 initial_state AgentState( messages[], questionLangChain和LangGraph的主要区别是什么, search_results[], decision ) final_state app.invoke(initial_state) print(\n--- 最终对话记录 ---) for msg in final_state.get(messages, []): print(f{type(msg).__name__}: {msg.content})预期效果这个图会开始运行。analyze节点首先判断需要search。然后执行search节点模拟搜索。搜索完成后流回analyze节点。此时analyze节点根据搜索到的结果可能决定answer或more_search。如果是answer则流向answer节点生成最终回复并结束。这个过程完美展示了循环和条件分支。6. 接口 API 与批量任务虽然 LangChain/LangGraph 本身不是 HTTP 服务但你可以轻松地将其核心逻辑封装成 API或用于处理批量任务。6.1 封装为 FastAPI 服务以下是将上述 LangGraph 智能体封装成 API 的示例# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from test_langgraph_agent import app as agent_workflow # 导入上面编译好的图 app FastAPI(titleLangGraph智能体API) class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str steps: list[str] app.post(/ask, response_modelQueryResponse) async def ask_question(request: QueryRequest): try: # 初始化状态并运行图 initial_state { messages: [], question: request.question, search_results: [], decision: } final_state agent_workflow.invoke(initial_state) # 提取最终答案 answer steps [] for msg in final_state.get(messages, []): if hasattr(msg, content): steps.append(msg.content) if 答案 in msg.content: answer msg.content.replace(答案, ).strip() return QueryResponse(answeranswer, stepssteps) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务后你就可以通过POST /ask接口提交问题并获得智能体分步思考后的答案。6.2 处理批量任务利用 LangGraph 的图结构处理批量问题非常高效。你可以并发或顺序地运行多个图实例。# batch_process.py from concurrent.futures import ThreadPoolExecutor from test_langgraph_agent import app as agent_workflow questions [ 什么是RAG, 如何用LangChain连接ChromaDB, LangGraph中的State有什么用, ] def process_one_question(q): 处理单个问题的函数 initial_state { messages: [], question: q, search_results: [], decision: } result agent_workflow.invoke(initial_state) # 提取答案逻辑... for msg in result.get(messages, []): if hasattr(msg, content) and 答案 in msg.content: return q, msg.content return q, 未找到答案 # 使用线程池并发处理注意如果LLM API有速率限制需调整 with ThreadPoolExecutor(max_workers3) as executor: futures [executor.submit(process_one_question, q) for q in questions] results [f.result() for f in futures] for question, answer in results: print(fQ: {question}) print(fA: {answer[:100]}...) # 打印前100字符 print(- * 50)7. 资源占用与性能观察LangChain 和 LangGraph 作为编排框架其本身的资源消耗CPU/内存可以忽略不计。性能瓶颈和资源占用的主体始终是你集成的 LLM和你调用的工具如向量数据库、外部 API。API 调用模式如果你使用 OpenAI GPT-4 等云端 API则主要消耗是网络延迟和 API 费用。本地几乎无计算资源占用。你需要监控 API 的响应时间和 Token 使用量。本地模型模式如果你通过 LangChain 集成 Ollama、 vLLM 或 Transformers 运行本地模型那么主要的资源占用就是模型加载和推理所需的GPU 显存和内存。观察显存在 Linux 可以使用nvidia-smi在 Windows 可以使用任务管理器性能选项卡查看 GPU 显存占用。性能关键点提示词长度过长的提示词会显著增加推理时间和内存消耗。图复杂度LangGraph 图中节点和边越多状态越复杂单次推理的循环次数可能越多导致总调用 LLM 的次数增加进而影响整体耗时。工具调用延迟如果图中包含搜索、数据库查询等外部工具这些工具的响应速度会极大影响工作流整体性能。优化建议对于复杂图使用streaming模式进行调试观察执行路径。为耗时长的工具调用设置超时timeout。在本地模型场景下合理设置模型的max_tokens和temperature参数以平衡速度与质量。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入 LangChain/LangGraph 失败1. 未安装2. 虚拟环境未激活3. Python 版本不兼容1.pip list查看是否安装2. 检查终端提示符3.python --version检查版本1. 重新安装pip install langchain langgraph2. 激活正确的虚拟环境3. 升级 Python 至 3.8调用 OpenAI API 超时或报错1. API Key 错误或未设置2. 网络问题3. 额度不足1. 检查代码中api_key2. 使用curl测试 API 连通性3. 登录 OpenAI 后台检查额度1. 设置正确的环境变量或代码 Key2. 检查代理或网络设置3. 充值或更换账户LangGraph 图编译或运行出错1.State结构定义错误2. 节点函数输入/输出与State不匹配3. 条件边conditional_edges路由函数返回了未定义的键1. 仔细检查TypedDict定义2. 打印节点函数的输入state检查其结构3. 调试路由函数确保返回值在预设映射中1. 修正State的类型注解2. 确保节点函数返回的字典能更新State3. 检查路由逻辑确保覆盖所有分支智能体陷入死循环图中存在未正确终止的循环使用app.get_graph().draw_mermaid()输出图结构可视化检查循环路径1. 在循环中增加条件判断如最大迭代次数2. 确保某个分支最终能流向END节点本地模型加载失败或显存不足1. 模型文件路径错误2. 显存小于模型所需1. 检查模型下载路径2. 使用nvidia-smi查看显存占用尝试加载更小模型1. 指定正确的模型路径2. 使用量化模型如 GGUF 格式或启用 CPU 卸载或升级硬件9. 最佳实践与使用建议从简开始逐步复杂不要一开始就设计庞大的图。先用 LangChain 的LCEL实现核心链路当发现需要复杂循环和状态管理时再迁移到 LangGraph。清晰定义状态在 LangGraph 中花时间设计好State结构。这是整个工作流的数据中枢清晰的类型提示TypedDict能避免后续很多调试麻烦。节点功能单一化每个节点函数只做一件事。例如一个节点专门调用 LLM另一个节点专门处理工具调用。这提高了代码的可读性和可测试性。善用可视化调试LangGraph 提供了get_graph().draw_mermaid()方法可以将图生成 Mermaid 图表。在复杂流程中可视化是理解与调试的利器。为生产环境做好准备错误处理在图的关键节点添加try...catch避免单个节点失败导致整个工作流崩溃。日志与监控记录每个节点的输入、输出和耗时便于追踪问题与性能分析。状态持久化对于长时间运行或需要中断恢复的智能体需要将State序列化存储到数据库或文件中。安全与合规工具权限严格控制智能体可以调用的工具特别是涉及写操作、网络访问或敏感信息的工具。输入输出过滤对用户输入和模型输出进行必要的清洗和过滤防止提示词注入或生成不当内容。数据留存根据规定妥善处理用户交互日志和生成内容。10. 总结与下一步LangChain 和 LangGraph 不是二选一的关系而是互补与进化的关系。LangChain 是你的“应用工厂”提供了构建 LLM 应用所需的所有标准化零件和装配线。LangGraph 则是这个工厂里的“智能流水线控制系统”当你需要生产具有复杂决策逻辑和状态记忆的“智能产品”时它提供了最高效的编排方式。对于初学者建议的路径是第一步掌握 LangChain 核心概念用LCEL搭建一个简单的 RAG 应用或聊天机器人。第二步当你的应用需要让 LLM 反复使用工具、进行多轮决策时开始学习 LangGraph。从官方最简单的 Agent 例子开始理解State和Graph的概念。第三步将你 LangChain 中笨重的多步 Chain 或 Agent 逻辑用 LangGraph 重新设计和实现体验其清晰和强大。最容易踩的坑是过早引入 LangGraph 带来的复杂度以及在图设计中状态管理混乱导致的调试困难。因此明确你的需求边界至关重要简单的线性流程用 Chain复杂的循环决策用 Graph。接下来你可以探索 LangGraph 更高级的特性如持久化检查点Checkpointer以实现智能体的长期记忆和中断恢复或者使用MessageGraph来构建更贴近多轮对话的架构。这个组合正在成为构建下一代复杂 AI 应用的事实标准框架。
RELATED — 相关阅读

相关资讯

LATEST — 最新资讯

最新发布

TODAY — 本日精选

新闻

WEEKLY — 本周精选

新闻

MONTHLY — 本月精选

新闻