从API调用到流程编排：LangChain核心概念与RAG实战指南-育师

1. 从“调用”到“编排”：我的LangChain认知转折点

刚开始接触LangChain的时候，我和很多人一样，以为它就是个“花架子”——一个把调用OpenAI API的几行代码包装得更复杂一点的库。那时候的想法很简单：大模型能力那么强，我直接发个请求，它给我返回结果不就完了？为什么还要引入一个中间层，学一堆新概念？这种想法，让我在很长一段时间里都只是在用LangChain的ChatOpenAI或LLMChain这类最基础的组件，本质上还是在做“一次性问答”。直到有一次，我试图构建一个能自动分析用户上传的PDF文档，并基于文档内容回答后续问题的系统时，才彻底撞了南墙。我发现，单纯地调用模型，完全无法处理“读取PDF -> 解析文本 -> 分割成块 -> 向量化存储 -> 根据问题检索相关片段 -> 组合成提示词 -> 生成最终答案”这一连串动作。我需要写大量的胶水代码来处理状态传递、错误处理、流程控制，代码很快就变成了一团乱麻。那一刻我才明白，LangChain的核心价值根本不是“调用”，而是“编排”。它提供了一套标准化的“乐高积木”（组件）和一套清晰的“搭建图纸”（Chain），让你能像组装流水线一样，将复杂任务拆解、串联、并行执行。写出第一条真正意义上的、能完成多步任务的Chain，就像第一次成功组装了一个能自动运行的机械装置，那种从混沌到有序的顿悟感，才是LangChain入门真正的开始。

2. 拆解“Chain”：它远不止是API的简单封装

很多人对Chain的误解，源于把它等同于一个函数调用链。但事实上，一个标准的LangChain Chain，其内涵要丰富得多。我们可以把它理解为一个有状态、可配置、可观测的工作单元。

2.1 Chain的三大核心构成：不止于顺序执行

一个功能完整的Chain，通常由三个关键部分有机组合而成，而不仅仅是几个LLM调用的前后拼接。

首先是组件。这是Chain的“肌肉”和“器官”。除了最显眼的LLM（大语言模型）和Prompt Templates（提示词模板），还包括：

文档加载器：从PDF、Word、网页、数据库等源头获取原始数据。
文本分割器：将长文档切割成模型上下文窗口能消化的小块，这里面的策略（按字符、按句子、按语义重叠）直接影响后续检索效果。
向量存储与检索器：将文本块转换为向量并存储，实现基于语义相似度的快速检索。这是实现RAG（检索增强生成）的基石。
输出解析器：将模型自由格式的文本输出，结构化地解析成Python对象（如Pydantic模型），方便下游程序处理。
工具：赋予模型“动手能力”，让它能执行搜索、计算、查数据库等具体操作，这是构建智能体（Agent）的基础。

其次是编排逻辑。这是Chain的“神经系统”和“骨架”。它定义了数据如何在组件间流动。最简单的LLMChain是线性串联：输入 -> 提示词填充 -> LLM调用 -> 输出。但复杂的Chain可能是：

条件分支：根据上一步的结果，决定下一步走哪条路径。
并行处理：同时处理多个子任务，然后合并结果。
循环迭代：对结果不满意时，自动调整参数重新执行，或对长列表进行分批处理。
状态传递与记忆：在多轮对话中，记住之前的交互历史，让模型拥有“上下文”。

最后是输入/输出模式。这是Chain的“接口”。一个设计良好的Chain，其invoke()或stream()方法的输入应该是一个结构清晰的字典，输出也应该是一个定义明确的格式。这保证了Chain本身可以作为更复杂流程中的一个可靠组件被调用。例如，一个问答Chain的输入可能是{“question”: “...”, “chat_history”: [...]}，输出是{“answer”: “...”, “source_documents”: [...]}。这种标准化使得组合和调试变得可能。

注意：很多初学者会忽略输出解析器。直接使用模型的原始字符串输出，会让后续处理非常脆弱。定义一个Pydantic模型来规范输出格式，是提升Chain鲁棒性的关键一步。

2.2 从“调用链”到“Chain”的思维转变

为了更直观地理解，我们对比一下两种思维模式下的代码。假设任务是将用户问题翻译成英文，然后用英文问题去查询一个数据库。

旧思维（胶水代码式调用链）：

# 伪代码，示意混乱的状态管理 def old_way(question): # 步骤1：翻译 prompt1 = f"将以下中文问题翻译成英文：{question}" response1 = openai_chat(prompt1) # 需要手动解析response1里的英文文本，可能还要处理错误 english_question = parse_response(response1) # 步骤2：查询 # 需要手动拼接查询语句，处理数据库连接和异常 sql = f"SELECT * FROM kb WHERE content LIKE '%{english_question}%'" result = database_query(sql) # 步骤3：组合最终答案 final_prompt = f"基于以下信息：{result}， 用中文回答原问题：{question}" final_answer = openai_chat(final_prompt) return final_answer

这段代码的问题显而易见：错误处理分散、中间状态（english_question,result）需要手动维护、流程僵化难以复用或修改。

新思维（LangChain Chain编排）：

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI from langchain.schema import Document from langchain.chains import SequentialChain # 定义子Chain 1：翻译 translation_prompt = ChatPromptTemplate.from_template( "将以下中文问题翻译成英文：{input}" ) translate_chain = translation_prompt | ChatOpenAI(model="gpt-4") | StrOutputParser() # 定义子Chain 2：查询（这里用模拟函数代替真实数据库工具） def retrieve_documents(question: str) -> list[Document]: # 这里可以是向量检索、数据库查询等 return [Document(page_content=f"关于'{question}'的模拟知识内容。")] # 定义子Chain 3：生成最终答案 qa_prompt = ChatPromptTemplate.from_template( "基于以下上下文：{context}， 用中文回答原问题：{original_question}" ) qa_chain = qa_prompt | ChatOpenAI(model="gpt-4") | StrOutputParser() # 编排主Chain overall_chain = { # 第一步：翻译，输出结果键名为“english_question” "english_question": translate_chain, # 保留原始输入，传递给后续步骤 "original_question": lambda x: x["input"], } | { # 第二步：检索，使用上一步的“english_question”作为输入 "context": lambda x: retrieve_documents(x["english_question"]), "original_question": lambda x: x["original_question"], } | { # 第三步：生成答案，使用“context”和“original_question” "output": qa_chain } # 调用 result = overall_chain.invoke({"input": "LangChain是什么？"}) print(result["output"])

这个Chain的清晰之处在于：数据流显式定义（通过字典键名传递），每个步骤职责单一且可测试，整体结构可复用、可扩展。如果你想在翻译后增加一个步骤来修正语法，只需要在编排中插入一个新的子Chain即可，无需重写整个逻辑。这就是编排思维带来的模块化优势。

3. 实战：构建你的第一条“真正”的Chain——一个简易RAG问答系统

理论说再多，不如亲手搭一个。我们来构建一个虽然简易但五脏俱全的RAG问答Chain。这个Chain将完成：加载本地PDF -> 分割文本 -> 创建向量索引 -> 接收用户问题 -> 检索相关文本 -> 生成答案。你会完整地看到多个组件是如何被串联起来的。

3.1 环境准备与核心组件选型

首先，你需要安装必要的包。这里我们选择比较通用和稳定的组合。

pip install langchain langchain-openai langchain-community pypdf chromadb tiktoken

langchain: 核心框架。
langchain-openai: OpenAI模型集成。
langchain-community: 包含大量第三方集成（如文档加载器）。
pypdf: 用于读取PDF文件。
chromadb: 一个轻量级的本地向量数据库。
tiktoken: 用于文本分割时精确计算Token。

接下来，我们初始化关键组件。假设你有一个名为knowledge.pdf的文件放在项目根目录。

import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_chroma import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 1. 初始化模型和嵌入模型 # 请将your_api_key替换为你的OpenAI API Key，或通过环境变量OPENAI_API_KEY设置 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 2. 加载并分割文档 loader = PyPDFLoader("knowledge.pdf") documents = loader.load() # 文本分割器：这里采用递归字符分割，保证句子完整性，并设置重叠以减少信息割裂 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 每个块约1000字符 chunk_overlap=200, # 块之间重叠200字符 length_function=len, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) all_splits = text_splitter.split_documents(documents) print(f"原始文档页数：{len(documents)}， 分割后块数：{len(all_splits)}")

这里有几个关键选择需要解释：

为什么用RecursiveCharacterTextSplitter？因为它尝试按字符顺序（从大段落到小词语）进行分割，能更好地保持语义段落完整性，比简单的按固定长度切割更智能。
chunk_size和chunk_overlap怎么定？chunk_size需要小于模型上下文窗口，并预留出问题和答案的空间。1000是一个常用起点。chunk_overlap用于避免一个完整的句子或概念被硬生生切到两个块里，导致检索时信息不完整，通常设为chunk_size的10%-20%。
为什么初始化两个模型？ChatOpenAI用于生成，OpenAIEmbeddings用于将文本转换为向量。它们是不同的API端点，虽然可以共用同一个基础模型系列，但职责不同。

3.2 构建向量数据库与检索链

分割好的文本块需要被转换成向量（一组数字），并存储起来，以便后续根据问题快速找到最相关的文本块。

# 3. 创建向量存储（向量数据库） # persist_directory指定持久化目录，这样下次运行无需重新生成向量 vectorstore = Chroma.from_documents( documents=all_splits, embedding=embeddings, persist_directory="./chroma_db" # 数据将保存在本地`chroma_db`文件夹 ) # 创建检索器。search_kwargs中的`k`值决定了返回多少个最相关的文档片段。 retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # 我们可以立即测试一下检索器 test_question = "LangChain的主要用途是什么？" docs = retriever.invoke(test_question) print(f"检索到 {len(docs)} 个相关片段:") for i, doc in enumerate(docs): print(f"[片段{i+1}] {doc.page_content[:200]}...") # 打印前200字符

这一步完成后，你的本地chroma_db文件夹里就存储了所有文本块的向量索引。下次程序启动时，你可以用Chroma(persist_directory=“./chroma_db”, embedding_function=embeddings)直接加载，无需再次计算嵌入，节省时间和API费用。

3.3 组装完整的RAG Chain

现在，我们将检索器、提示词模板和LLM组装成一条完整的Chain。这是最体现“编排”思想的一步。

# 4. 定义提示词模板 # 这是一个标准的RAG提示词模板，明确指令模型基于提供的上下文回答问题。 template = """你是一个专业的助手，请严格根据以下提供的上下文信息来回答问题。 如果你不知道答案，就诚实地回答你不知道，不要编造信息。 上下文： {context} 问题：{question} 请根据上下文给出答案：""" prompt = ChatPromptTemplate.from_template(template) # 5. 组装RAG Chain # 这是Chain的核心编排逻辑 rag_chain = ( # 第一步：准备输入。`RunnablePassthrough()`用于传递原始的`question`。 # `retriever`会接收`question`，并检索出相关的文档，赋值给`context`。 {"context": retriever, "question": RunnablePassthrough()} # 第二步：填充提示词。上一步的输出（包含context和question）会填充到prompt模板中。 | prompt # 第三步：调用大语言模型生成答案。 | llm # 第四步：解析输出，将模型的`AIMessage`对象转换为纯文本字符串。 | StrOutputParser() ) # 6. 调用Chain question = "使用LangChain构建应用有哪些优势？" answer = rag_chain.invoke(question) print(f"问题：{question}") print(f"答案：{answer}")

这个rag_chain的定义非常清晰，它用|操作符（LangChain的LCEL语法）描述了数据流：检索上下文 + 问题 -> 填充提示词 -> 模型生成 -> 解析输出。RunnablePassthrough()是一个特殊的组件，它不做任何处理，只是将输入原封不动地传递下去，在这里用于确保question这个字段能一路传递到prompt模板。

3.4 进阶：为Chain增加历史对话能力

上面的Chain是单轮的。在实际对话中，我们需要让模型记住之前说过的话。这就需要引入“记忆”组件。LangChain提供了多种记忆后端，这里我们使用最简单的ConversationBufferMemory。

from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationalRetrievalChain # 重新初始化一个带记忆的检索器Chain memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) # 使用LangChain内置的高级Chain：ConversationalRetrievalChain # 它内部帮我们处理了历史对话与当前问题的拼接逻辑。 conversational_chain = ConversationalRetrievalChain.from_llm( llm=llm, retriever=retriever, # 使用我们之前创建好的检索器 memory=memory, verbose=True, # 设置为True可以看到Chain运行的详细步骤，便于调试 combine_docs_chain_kwargs={"prompt": prompt} # 可以传入自定义的提示词 ) # 进行多轮对话 result1 = conversational_chain.invoke({"question": "LangChain是什么？"}) print(f"AI: {result1['answer']}") # 第二问可以指代上文 result2 = conversational_chain.invoke({"question": "它和LangGraph有什么区别？"}) print(f"AI: {result2['answer']}") # 查看记忆内容 print(memory.load_memory_variables({}))

ConversationalRetrievalChain是一个更高级的封装，它自动将chat_history和当前question组合成一个新的、包含完整上下文的问题，再用这个问题去检索和生成。当你设置verbose=True时，控制台会打印出内部步骤，这对于理解Chain的执行过程和调试至关重要。

4. 调试与优化：让Chain从“能跑”到“好用”

写出第一条能运行的Chain只是起点。接下来你会遇到各种问题：答案胡言乱语、检索不到相关内容、速度太慢等等。这时就需要深入的调试和优化。

4.1 诊断Chain问题的“三板斧”

当Chain输出不符合预期时，不要盲目修改代码，应该系统性地排查。

第一板斧：检查输入输出。使用invoke时，在每个步骤间插入打印语句，或者使用LangChain的RunnableLambda来包装函数以打印中间值。对于上面定义的rag_chain，你可以拆开检查：

# 拆解检查 from langchain_core.runnables import RunnableLambda def debug_print(x): print(f"[DEBUG] 步骤输入: {x}") return x debug_chain = ( {"context": retriever, "question": RunnablePassthrough()} | RunnableLambda(debug_print) # 查看检索到的上下文 | prompt | RunnableLambda(lambda x: print(f"[DEBUG] 填充后的提示词:\n{x.to_string()}") or x) # 查看发送给模型的完整提示词 | llm | RunnableLambda(lambda x: print(f"[DEBUG] 模型原始响应: {x}") or x) | StrOutputParser() ) # 调用debug_chain，观察控制台输出

第二板斧：优化检索质量。RAG效果不佳，十有八九是检索环节出了问题。

检索不到？检查retriever.invoke(question)返回的文档是否真的与问题相关。可能是嵌入模型不适合你的领域，或者chunk_size设置不当导致信息碎片化。可以尝试换用其他嵌入模型（如text-embedding-3-large），或者调整分割策略。
检索到但没用？可能是k值太小，没有包含关键信息；也可能是检索到的片段质量差。可以尝试：
- 增加k值，比如从4调到8。
- 使用MMR（最大边际相关性）检索，在保证相关性的同时增加多样性，避免返回内容过于同质。
```
retriever = vectorstore.as_retriever( search_type="mmr", # 使用MMR算法 search_kwargs={"k”: 8, “fetch_k”: 20} # 最终返回8个，从最相关的20个中筛选 )
```
- 在提示词中加强指令，明确要求模型“如果上下文不相关，请忽略”。

第三板斧：优化提示词工程。模型的输出质量极大程度依赖于提示词。

指令不清晰：确保你的提示词模板包含了明确的任务指令、角色设定和输出格式要求。
上下文位置不当：对于某些模型，将上下文放在问题前面还是后面效果可能不同，可以尝试调整模板结构。
加入“少样本示例”：在提示词中提供一两个输入输出的例子，能显著提升模型在复杂任务上的表现。

better_template = """你是一个严谨的技术专家。请根据提供的上下文回答问题。 如果上下文包含答案，请精确引用。如果上下文不包含足够信息，请说“根据已有信息无法回答”。 示例： 上下文：苹果是一种水果，富含维生素C。 问题：苹果有什么营养？ 答案：根据上下文，苹果富含维生素C。 现在请回答真实问题： 上下文：{context} 问题：{question} 答案："""

4.2 性能与成本考量

Chain跑起来后，你需要关注它的效率和花费。

异步调用：如果Chain中有多个可以并行执行的独立步骤（比如同时检索多个不同来源的数据），使用ainvoke或abatch可以大幅缩短耗时。
缓存：对于重复的、确定性的计算（如对相同文本块的嵌入计算），启用缓存可以节省大量成本和时间。LangChain支持内存缓存(InMemoryCache)或SQLite缓存(SQLiteCache)。
```
from langchain.globals import set_llm_cache from langchain.cache import SQLiteCache set_llm_cache(SQLiteCache(database_path=".langchain.db"))
```
流式输出：对于需要长时间生成的回答，使用stream()方法可以实现逐词输出，提升用户体验。
```
for chunk in rag_chain.stream(“请详细解释Chain的概念”): print(chunk, end="", flush=True)
```
Token计数与成本估算：使用tiktoken或模型的get_num_tokens方法，在调用前估算提示词的Token数量，对成本做到心中有数。特别是在处理长文档时，检索返回的上下文总长度是成本的主要变量。

5. 超越简单Chain：Agent与LangGraph的初探

当你熟练掌握了Chain的编排，你会发现有些问题用固定的流程无法完美解决。比如，用户的问题可能需要先搜索网络，再查数据库，最后进行推理计算。步骤的顺序和数量在运行前无法确定。这时，你就需要更强大的范式：智能体。

5.1 从Chain到Agent：赋予模型决策权

智能体的核心思想是将LLM作为决策中心，它根据当前目标和可用工具，动态决定下一步该做什么。你可以把它看作一个由LLM驱动的、可自主调用工具的超级Chain。

from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain.tools import Tool from langchain import hub # 1. 定义工具。工具是Agent可以调用的函数。 # 假设我们有两个工具：一个用于计算，一个用于检索我们之前构建的向量库。 from langchain.tools import tool import math @tool def calculator(expression: str) -> str: """计算一个数学表达式的值。例如：‘calculator(“3 + 5 * 2”)’""" try: # 警告：使用eval有安全风险，仅作演示。生产环境应用更安全的方法。 result = eval(expression, {"__builtins__": {}}, math.__dict__) return str(result) except Exception as e: return f"计算错误：{e}" # 将之前的检索器包装成工具 retriever_tool = Tool( name="knowledge_base", func=retriever.invoke, # 直接使用我们之前定义的检索器 description="当需要询问关于LangChain、AI编程或本项目文档的具体知识时，使用此工具。" ) tools = [calculator, retriever_tool] # 2. 拉取一个预设的Agent提示词（来自LangChain Hub） prompt = hub.pull("hwchase17/openai-tools-agent") # 3. 创建Agent agent = create_openai_tools_agent(llm, tools, prompt) # 4. 创建Agent执行器 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 5. 运行Agent result = agent_executor.invoke({ "input": “请先告诉我LangChain是什么，然后计算25的平方根是多少？” }) print(result["output"])

运行这段代码，并观察verbose=True时的日志，你会看到类似这样的思考过程：

> 进入新的AgentExecutor链... 思考：用户问了两个问题。第一个是关于LangChain定义的，我需要使用knowledge_base工具。第二个是数学计算，需要使用calculator工具。 行动：调用knowledge_base工具，参数：{"input": "LangChain是什么"} 观察：[检索到的文档内容...] 思考：根据知识库，我得到了LangChain的定义。现在回答第一个问题，然后处理第二个计算问题。 行动：调用calculator工具，参数：{"expression": "math.sqrt(25)"} 观察：5.0 思考：我已回答了第一个问题，并计算出了25的平方根是5.0。现在可以给出最终答案了。 最终答案：LangChain是一个用于开发由大语言模型驱动的应用程序的框架... 25的平方根是5.0。

这就是Agent的魅力：模型自己规划了步骤顺序（先检索，后计算），并选择了正确的工具。你不再需要手动编写“如果问题是A则做X，如果是B则做Y”的逻辑分支。

5.2 LangGraph：当流程需要循环与状态

SequentialChain和普通的Agent适合线性或简单分支的任务。但对于需要多轮交互、循环（比如反复修正一个方案直到满意）、或者复杂状态管理的任务，我们就需要LangGraph。它允许你用图（Graph）的方式来定义工作流，节点是组件或工具，边定义了运行路径。

一个经典用例是递归式写作助手：生成大纲 -> 撰写章节 -> 检查质量 -> 如不达标，返回修改。

# 注：这是一个概念性示例，展示LangGraph的循环控制思想。 from langgraph.graph import StateGraph, END from typing import TypedDict # 定义状态结构 class WritingState(TypedDict): topic: str outline: str draft: str feedback: str revision_count: int # 定义各个节点函数（这里用伪代码） def generate_outline(state: WritingState): # 调用LLM生成大纲 state[“outline”] = llm.invoke(f”为‘{state[‘topic’]}’生成写作大纲”) return state def write_draft(state: WritingState): # 根据大纲撰写初稿 state[“draft”] = llm.invoke(f”根据大纲写作：{state[‘outline’]}”) return state def review_draft(state: WritingState): # 检查初稿质量 state[“feedback”] = llm.invoke(f”评审以下草稿：{state[‘draft’]}。给出修改建议。”) state[“revision_count”] += 1 return state def should_continue(state: WritingState): # 决策节点：根据评审意见和修改次数决定继续还是结束 if “需要重写” in state[“feedback”] and state[“revision_count”] < 3: return “rewrite” # 返回下一个节点的名称 else: return END # 构建图 workflow = StateGraph(WritingState) workflow.add_node(“generate_outline”, generate_outline) workflow.add_node(“write_draft”, write_draft) workflow.add_node(“review_draft”, review_draft) # 设置边 workflow.set_entry_point(“generate_outline”) workflow.add_edge(“generate_outline”, “write_draft”) workflow.add_edge(“write_draft”, “review_draft”) # 条件边：根据`should_continue`函数的返回值决定走向 workflow.add_conditional_edges( “review_draft”, should_continue, {“rewrite”: “write_draft”, END: END} # 如果返回”rewrite”，则跳回”write_draft”节点 ) # 编译并运行图 app = workflow.compile() initial_state = {“topic”: “如何学习LangChain”, “revision_count”: 0} final_state = app.invoke(initial_state)

在这个图中，review_draft节点后的条件边形成了一个循环：如果评审认为需要重写且次数未超限，流程就会回到write_draft节点。这种带有循环和状态依赖的复杂流程，用传统的Chain很难优雅地实现，而用LangGraph则非常直观。

所以，回到标题的感悟：“调用”是单次操作，“Chain”是固定流水线，“Agent”是自主任务分解，“Graph”是可控的复杂工作流。LangChain提供的是一套逐步升级的武器，让你能应对从简单到极其复杂的AI应用开发需求。写出第一条Chain，只是你拿到了第一把钥匙，门后还有更广阔的世界等着你去构建。