LangChain与LangGraph实战:从RAG到智能体的AI应用开发指南
2026/8/4 2:11:32 网站建设 项目流程

1. 先搞清楚 LangChain 和 LangGraph 到底能帮你解决什么问题

如果你正在找一套能直接上手、把大语言模型(LLM)和你的数据、工具、业务流程结合起来的方案,那 LangChain 和 LangGraph 就是目前最值得投入时间学习的框架之一。它们不是另一个需要你从零搭建的 AI 模型,而是一个“连接器”和“编排器”,核心价值在于帮你把 OpenAI、Anthropic、本地模型这些 LLM 能力,和你自己的文档、数据库、API 接口、乃至复杂的多步骤任务逻辑,高效、稳定地串联起来。

很多人一上来就被“RAG”、“智能体”这些术语吓住,或者陷入到无穷无尽的 API 调用细节里。其实,LangChain 要解决的核心痛点非常具体:当你有一个强大的 LLM,但需要它基于你的私有数据回答问题(RAG),或者需要它按特定流程调用工具完成任务(智能体)时,LangChain 提供了一套标准化的、可复用的组件和模式,让你不用重复造轮子。而 LangGraph 则是在 LangChain 基础上,专门为构建有状态、可循环、多角色协作的复杂智能体系统而设计的。

所以,这套教程的价值不在于教你最前沿的 AI 理论,而在于提供一条从零到一构建可用应用的清晰路径。它适合已经了解 Python 基础、对 LLM API 有初步接触(比如调用过 ChatGPT API),但不知道如何将其工程化、产品化的开发者、产品经理或技术爱好者。学完之后,你应该能独立搭建一个简单的文档问答机器人,或者设计一个能自动执行多步骤任务的智能体流程。

2. 学习前的环境准备与核心依赖确认

在开始跟着任何教程敲代码之前,先把环境理顺,能避免至少一半的“跑不通”问题。LangChain 生态迭代很快,但核心依赖相对稳定。

基础环境要求:

  • Python: 推荐使用 Python 3.8 到 3.11 版本。3.12 及以上版本可能存在一些第三方库的兼容性问题,新手建议先避开。
  • 包管理工具: 强烈建议使用pip配合virtualenvconda创建独立的虚拟环境。这是保证项目依赖不冲突的最佳实践。
  • 代码编辑器: VS Code 或 PyCharm 均可,确保有好的 Python 插件支持。

核心依赖安装:打开终端,在你的项目虚拟环境中,执行以下命令安装最核心的包:

pip install langchain langchain-community langchain-core

这行命令安装了 LangChain 的核心框架、社区贡献的第三方集成以及核心抽象。这是构建大多数应用的基础。

关键环境变量(API Keys):LangChain 本身不提供模型,你需要接入一个 LLM 服务。对于学习和快速验证,OpenAI 的 API 是最常见的选择。

  1. 前往 OpenAI 平台注册并获取 API Key。
  2. 在命令行中临时设置环境变量(每次新开终端都需要):
    export OPENAI_API_KEY="你的-api-key"
  3. 或者在项目根目录创建.env文件,写入OPENAI_API_KEY=你的-api-key,并使用python-dotenv包在代码中加载。这是更安全、更工程化的做法。
pip install python-dotenv openai

验证安装:创建一个简单的test_env.py文件,写入以下代码:

import os from langchain_openai import ChatOpenAI from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 llm = ChatOpenAI(model="gpt-3.5-turbo") # 使用 gpt-3.5-turbo 验证,成本更低 response = llm.invoke("你好,请用一句话介绍你自己。") print(response.content)

如果能正常收到 LLM 的回复,说明基础环境和 API 连接成功。如果报错,优先检查:

  1. OPENAI_API_KEY是否正确设置且有效。
  2. 网络连接是否正常,能否访问 OpenAI API。
  3. Python 和 pip 版本是否匹配。

3. 构建你的第一个 RAG 应用:从文档加载到智能问答

RAG(检索增强生成)是 LangChain 最经典的应用场景。它的流程可以简化为:加载你的文档 -> 切分成片段 -> 转换成向量并存储 -> 提问时检索相关片段 -> 连同问题和片段一起交给 LLM 生成答案。下面我们拆解每一步。

3.1 文档加载与文本分割

LangChain 提供了大量的DocumentLoader,支持从 TXT、PDF、PPT、网页、Notion 等来源加载文档。这里以本地 TXT 文件为例。

pip install pypdf # 如果你需要处理 PDF,安装这个
from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 1. 加载文档 loader = TextLoader("./your_document.txt", encoding="utf-8") documents = loader.load() # 2. 分割文本 # 直接使用默认分割器可能不合适,需要根据文档特点调整。 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个片段的最大字符数 chunk_overlap=50, # 片段之间的重叠字符数,保持上下文连贯 separators=["\n\n", "\n", "。", "!", "?", ",", " ", ""] # 分割符优先级 ) split_docs = text_splitter.split_documents(documents) print(f"原始文档被切分为 {len(split_docs)} 个片段。")

关键参数解析:

  • chunk_size: 太小会丢失上下文,太大会降低检索精度并增加 LLM 处理负担。对于通用文本,500-1000 是个不错的起点。
  • chunk_overlap: 防止一个句子或关键信息被硬生生切断。通常设为chunk_size的 10%-20%。
  • separators: 定义了分割的优先级。这里的意思是先按双换行分,不行再按单换行分,再按句号分……这样能尽可能在语义边界处切割。

3.2 向量化与向量数据库存储

文本片段需要转换成计算机能理解的“向量”(一组数字),这个过程叫嵌入(Embedding)。然后存入向量数据库,以便后续快速检索。

pip install chromadb langchain-openai tiktoken # Chroma 是一个轻量级、开源的向量数据库,非常适合学习和原型开发。
from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 1. 初始化嵌入模型 # 使用 OpenAI 的 text-embedding-ada-002,注意它和 Chat 模型是分开计费的。 embeddings = OpenAIEmbeddings(model="text-embedding-ada-002") # 2. 将分割后的文档转换为向量并存入 Chroma # persist_directory 指定向量数据库持久化到磁盘的路径 vectorstore = Chroma.from_documents( documents=split_docs, embedding=embeddings, persist_directory="./chroma_db" # 数据将保存在这个目录 ) vectorstore.persist() # 显式持久化 print("向量数据库已创建并保存。")

重要提醒:

  • 嵌入模型和聊天模型是独立的。OpenAIEmbeddings调用会产生额外的 API 费用。
  • Chroma将向量数据保存在本地./chroma_db目录。首次运行会创建,后续可以直接加载,无需重新生成向量,除非文档更新。
  • 生产环境可能会考虑PineconeWeaviate等托管服务,但本地 Chroma 对于中小规模数据完全够用。

3.3 构建检索链并进行问答

现在,我们已经有了一个“知识库”。接下来构建一个链条:用户提问 -> 从向量库检索相关片段 -> 组合成提示词 -> 发送给 LLM -> 返回答案。

from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI # 1. 从磁盘加载已存在的向量数据库 vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embeddings ) # 2. 将向量数据库转换为一个检索器 retriever = vectorstore.as_retriever( search_type="similarity", # 相似度搜索 search_kwargs={"k": 3} # 返回最相关的 3 个片段 ) # 3. 创建 LLM llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # temperature=0 使输出更确定 # 4. 创建检索问答链 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 最常用的类型,将所有检索到的文档“塞”进上下文 retriever=retriever, return_source_documents=True # 返回参考来源,便于验证 ) # 5. 进行提问 question = "根据文档,XX项目的核心目标是什么?" result = qa_chain.invoke({"query": question}) print(f"问题:{question}") print(f"答案:{result['result']}") print("\n--- 参考来源 ---") for doc in result['source_documents']: print(f"内容片段:{doc.page_content[:200]}...") # 打印前200字符 print(f"来源:{doc.metadata}\n")

链条类型chain_type选择:

  • "stuff": 最简单直接,将所有检索到的文档内容合并后一次性发送给 LLM。适用于检索片段总长度不超过 LLM 上下文窗口的情况。
  • "map_reduce": 先让 LLM 对每个片段单独总结,再对总结进行归纳。适合处理大量文档,但调用 API 次数多,速度慢。
  • "refine": 迭代式处理,用上一个片段的答案来完善下一个片段的答案。质量可能更高,但更慢。
  • "map_rerank": 对每个片段打分并排序,只选用高分片段。需要支持打分的模型。

对于大多数入门和中等复杂度场景,"stuff"是首选。你需要确保chunk_size * k不超过 LLM 的上下文限制(如 GPT-3.5-turbo 是 16K tokens)。

4. 进阶到智能体:用 LangChain 让 LLM 学会使用工具

RAG 解决了“知识”问题,智能体(Agent)则要解决“行动”问题。智能体的核心思想是:LLM 作为“大脑”,根据用户请求和当前状态,决定下一步是直接回答,还是调用某个工具(如搜索、计算、查询数据库)来获取信息,然后继续思考,直到得出最终答案。

4.1 定义工具

工具可以是任何可执行的函数,它接收文本输入,返回文本输出。LangChain 要求用@tool装饰器来声明。

from langchain.agents import tool import requests from datetime import datetime @tool def get_current_time(tz: str = "Asia/Shanghai") -> str: """获取指定时区的当前时间。输入应为时区名称,例如 'Asia/Shanghai' 或 'UTC'。""" # 这是一个简化示例,实际应使用 pytz 或 zoneinfo 库 now = datetime.now() return f"The current time in {tz} is approximately {now.strftime('%Y-%m-%d %H:%M:%S')}." @tool def search_web(query: str) -> str: """使用搜索引擎(示例用 DuckDuckGo)搜索网络信息。输入应为搜索关键词。""" # 注意:实际使用需要安装 duckduckgo-search 库并处理可能的不稳定 try: from duckduckgo_search import DDGS with DDGS() as ddgs: results = list(ddgs.text(query, max_results=3)) return "\n".join([f"{r['title']}: {r['body']}" for r in results]) except ImportError: return "Error: Please install 'duckduckgo-search' package first." except Exception as e: return f"Search error: {str(e)}" # 将工具放入列表,供智能体使用 tools = [get_current_time, search_web]

每个工具都必须有清晰的名称描述。LLM 正是通过这些描述来理解何时以及如何使用该工具。描述要尽可能准确。

4.2 创建智能体执行器

我们需要一个“执行器”来协调 LLM 的思考、工具调用和结果整合。这里使用 OpenAI 函数调用(Function Calling)作为智能体的底层机制,这是目前最稳定、高效的方式。

from langchain_openai import ChatOpenAI from langchain.agents import create_openai_functions_agent, AgentExecutor from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder # 1. 创建 LLM llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 2. 创建提示词模板 # SYSTEM_MESSAGE 定义了智能体的角色和能力 SYSTEM_MESSAGE = """你是一个有用的助手,可以访问以下工具: {tools} 使用这些工具来回答用户的问题。如果你不需要使用工具,也可以直接回答。 请始终以中文回复。""" prompt = ChatPromptTemplate.from_messages([ ("system", SYSTEM_MESSAGE), MessagesPlaceholder(variable_name="chat_history"), # 预留对话历史的位置 ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 智能体思考过程 ]) # 3. 创建智能体 agent = create_openai_functions_agent(llm=llm, tools=tools, prompt=prompt) # 4. 创建执行器 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 设置为 True 可以看到详细的思考步骤,调试时非常有用 handle_parsing_errors=True, # 处理解析错误,避免因格式问题直接崩溃 max_iterations=5 # 限制最大迭代次数,防止死循环 )

4.3 运行智能体并观察其思考过程

verbose=True后运行,你会在控制台看到类似下面的输出,这是理解智能体如何工作的关键。

question = "上海现在几点了?顺便搜索一下今天的热点新闻。" result = agent_executor.invoke({"input": question, "chat_history": []}) print("\n--- 最终答案 ---") print(result["output"])

控制台输出示例(verbose 模式):

> Entering new AgentExecutor chain... Thought: 用户问了两个问题:1. 上海当前时间。2. 今天的热点新闻。我需要使用工具来获取这些信息。 Action: get_current_time Action Input: {"tz": "Asia/Shanghai"} Observation: The current time in Asia/Shanghai is approximately 2024-05-27 15:30:45. Thought: 我已经得到了上海的时间。现在需要搜索今天的热点新闻。我需要使用搜索工具。 Action: search_web Action Input: {"query": "今日热点新闻 2024年5月27日"} Observation: [搜索返回的三条新闻摘要...] Thought: 我现在有了时间和新闻信息,可以综合起来回答用户了。 Action: Final Answer Final Answer: 上海现在是2024年5月27日下午3点30分左右。根据搜索,今日的热点新闻有:1. ...[新闻1摘要]。2. ...[新闻2摘要]。3. ...[新闻3摘要]。 > Finished chain.

通过verbose输出,你可以清晰地看到智能体的“思考-行动-观察”循环。这对于调试工具描述是否清晰、LLM 是否错误理解指令至关重要。

5. 构建复杂工作流:引入 LangGraph 实现多智能体与状态管理

当任务变得复杂,需要多个步骤循环、分支判断,或者多个“智能体”角色协作时,基础的AgentExecutor就显得力不从心。这时就需要LangGraph。它将工作流抽象为“图”(Graph),节点代表步骤(可以是工具调用、LLM调用或普通函数),边代表步骤之间的流转条件。

5.1 理解 LangGraph 的核心概念

  • State: 一个共享的字典,在整个工作流执行过程中传递和修改数据。这是 LangGraph 管理状态的核心。
  • Node: 节点,一个函数,接收当前 State,执行操作,并返回更新后的 State。
  • Edge: 边,决定下一个执行哪个 Node。可以是固定流转,也可以根据 State 中的某个值动态决定(条件边)。

5.2 构建一个简单的审阅工作流

假设我们有一个需求:用户提交一段文案,需要先后经过“拼写检查”和“风格优化”两个步骤。我们可以用 LangGraph 来编排。

from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END import operator # 1. 定义状态结构 class ReviewState(TypedDict): """工作流的状态定义""" original_text: str # 原始文案 spell_checked_text: str # 拼写检查后的文案 final_text: str # 最终优化后的文案 feedback: list[str] # 收集各步骤的反馈信息 # 2. 定义节点函数 def node_spell_check(state: ReviewState) -> ReviewState: """节点A:模拟拼写检查""" # 这里简化处理,实际可以调用专门的拼写检查API或库 text = state["original_text"] # 假设我们只是做个简单替换模拟检查 checked_text = text.replace("teh", "the").replace("adn", "and") feedback = f"拼写检查完成。修正了常见拼写错误。" return { "spell_checked_text": checked_text, "feedback": state["feedback"] + [feedback] } def node_style_optimize(state: ReviewState) -> ReviewState: """节点B:调用LLM进行风格优化""" from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7) prompt = f""" 请将以下文案优化得更专业、更吸引人。保持原意。 原文案:{state['spell_checked_text']} 优化后的文案: """ response = llm.invoke(prompt) optimized_text = response.content feedback = f"风格优化完成。LLM已生成优化版本。" return { "final_text": optimized_text, "feedback": state["feedback"] + [feedback] } # 3. 构建图 workflow = StateGraph(ReviewState) # 添加节点 workflow.add_node("spell_check", node_spell_check) workflow.add_node("style_optimize", node_style_optimize) # 设置边的连接关系:spell_check -> style_optimize -> END workflow.add_edge("spell_check", "style_optimize") workflow.add_edge("style_optimize", END) # 设置入口节点 workflow.set_entry_point("spell_check") # 编译图,得到可执行对象 app = workflow.compile()

5.3 执行工作流并查看结果

# 4. 初始化状态并执行 initial_state: ReviewState = { "original_text": "Our product is teh best in the market adn we are sure you will love it.", "spell_checked_text": "", "final_text": "", "feedback": [] } # 执行工作流 final_state = app.invoke(initial_state) print("原始文案:", final_state["original_text"]) print("\n拼写检查后:", final_state["spell_checked_text"]) print("\n最终优化文案:", final_state["final_text"]) print("\n工作流反馈:", final_state["feedback"])

这个例子展示了线性工作流。LangGraph 更强大的地方在于支持条件边循环。例如,你可以在style_optimize节点后,添加一个由 LLM 判断“文案是否足够好”的节点,如果不够好,就循环回style_optimize重新优化,直到满足条件或达到最大循环次数。这正是在构建复杂、动态的智能体系统时所需要的。

6. 实战避坑与生产化思考

跟着教程跑通 Demo 只是第一步。要把 LangChain/LangGraph 用于实际项目,以下几个坑点和优化方向必须提前考虑。

6.1 常见错误与排查顺序

  1. API 密钥或网络问题:任何与OpenAIAnthropic等相关的调用失败,首先检查环境变量OPENAI_API_KEY等是否设置正确,以及网络是否能正常访问对应 API 地址(对于国内用户,这可能是个常见问题)。
  2. 版本兼容性问题:LangChain 版本迭代快,某些接口或参数名可能会变。如果代码报ImportErrorAttributeError,第一反应是去查阅对应版本的官方文档(https://python.langchain.com/docs/),而不是盲目搜索。
  3. 提示词(Prompt)问题:LLM 输出不符合预期,比如不调用工具、格式错误。首先检查你的SYSTEM_MESSAGE和工具描述是否足够清晰。用verbose=True查看 LLM 的原始思考过程,往往能发现问题。
  4. 向量检索效果差:RAG 回答不准确。排查点:
    • 文本分割chunk_sizechunk_overlap是否合适?用print(split_docs)看看分割后的片段是否保持了语义完整性。
    • 检索策略search_kwargs={"k": 3}中的k值是否太小?可以尝试增加到 5 或 10。也可以试试search_type="mmr"(最大边际相关性),在相关性和多样性之间取得平衡。
    • 嵌入模型:对于中文场景,text-embedding-ada-002效果不错,但也可以尝试专门的多语言或中文嵌入模型。
  5. 智能体陷入死循环:智能体不停调用同一个工具。务必设置AgentExecutormax_iterations参数(如 10)。同时检查工具函数的返回值格式是否稳定,LLM 能否正确解析。

6.2 从 Demo 到生产环境的考量

  1. 异步与并发:LangChain 原生支持异步。对于需要处理大量请求的 Web 服务,务必使用ainvokeabatch等异步方法,并结合asyncio提高吞吐量。
    # 异步调用示例 async def process_question(question): result = await qa_chain.ainvoke({"query": question}) return result
  2. 缓存:频繁调用相同的 LLM 请求(例如相同的提示词)会产生不必要的费用和延迟。使用LangChain的缓存组件,如InMemoryCacheSQLiteCache,可以显著提升性能。
    from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())
  3. 日志与监控:生产系统必须要有完善的日志。记录每一次 LLM 调用、工具调用的输入输出、耗时和 Token 使用量。这有助于成本核算、问题排查和效果优化。
  4. 错误处理与重试:网络波动、API 限流不可避免。使用tenacity等库为 LLM 和工具调用添加重试机制。在AgentExecutor中,利用handle_parsing_errors等参数优雅处理部分错误。
  5. 向量数据库的选择:本地 Chroma 适合原型和中小数据量。当数据量极大(百万级以上)或需要高可用、分布式时,需要考虑PineconeWeaviateQdrant等专业向量数据库服务。
  6. 提示词工程与管理:不要将提示词硬编码在代码中。考虑将其外置到配置文件、数据库或专门的提示词管理平台,便于迭代和 A/B 测试。

6.3 学习路径建议

不要试图一次性掌握 LangChain 的所有模块。建议按以下路径循序渐进:

  1. 核心概念Model I/O(Prompt/LLM/OutputParser),Data Connection(Document Loader/Text Splitter/Vectorstore),Chains(LLMChain, SequentialChain)。
  2. 重点突破:深入掌握RetrievalQA链和OpenAI Functions Agent,这是使用频率最高的两部分。
  3. 探索生态:根据需求学习LangChain Community里的各种集成,如邮件工具、SQL 数据库工具、GitHub 工具等。
  4. 进阶编排:当遇到需要循环、分支、多角色协作的复杂场景时,再深入学习LangGraph
  5. 关注官方动态:LangChain 生态发展迅速,关注其官方博客和 Discord,了解新特性(如 LangSmith 用于跟踪和评估, LangServe 用于部署)的最佳实践。

最终,评判你是否掌握了 LangChain,不是背下了多少 API,而是能否独立设计并实现一个解决实际问题的、健壮的 AI 应用流程。从用一个清晰的提示词驱动 LLM,到用 Chain 串联多个步骤,再到用 Agent 动态决策,最后用 Graph 管理复杂状态,这条路径上的每一步,都对应着真实项目中不断增长的需求复杂度。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询