LangChain 核心基础：从零搭建你的单线大模型工作流（实战篇）

星野暗涌2026-05-082026-07-04

一、引言：为什么我们需要 LangChain？

在刚接触大语言模型（LLM）开发时，很多人觉得写个 API 请求就万事大吉了。然而，随着业务深入，原生大模型开发的四大痛点会迅速浮现：

产生幻觉：简单的提示词往往无法约束模型，导致它在没有知识储备时“一本正经地胡说八道”。
输出非结构化：模型天生喜欢输出自然语言的对话，但我们的代码接口（如前端渲染或数据库插入）通常需要严格的 JSON 格式，两者难以直接对接。
知识陈旧且缺乏实时性：模型只停留在训练数据的截止日期前，不知道今天的天气，也不知道公司昨天刚发布的内部文件。
缺乏行动力：原生大模型是一个“被锁在小黑屋里的高智商大脑”，无法主动联网搜索、操作计算器或读写数据库。

LangChain 的破局之道在于其“积木化”的设计哲学。它将复杂的自然语言处理（NLP）流程拆解为一系列标准化的组件（如语言模型、提示词模板、输出解析器、检索器等）。借助 LangChain，开发者可以像搭积木一样自由组合这些组件，高效定制出强大且稳定的工作流。

二、核心基石：Runnable 接口与 LCEL 语法

要搞懂 LangChain，就必须理解它底层最核心的两个概念：Runnable 与 LCEL。

1. 统一的 Runnable 接口

在 LangChain 中，万物皆可“运行”。无论是模型、解析器还是检索器，几乎所有核心组件都实现了一个统一的 Runnable 接口。这意味着你可以用完全一致的方法来调用它们：

invoke()：单次调用，输入一个数据，等待并返回一个结果。
stream()：流式传输，常用于打字机效果，逐步输出生成的内容。
batch()：批处理，一次性传入多个输入，并行处理以提高效率。

2. LangChain 表达式语言 (LCEL)

LCEL（LangChain Expression Language） 是一种声明式的语法，也是 LangChain 框架最大的魔法。

它巧妙地重载了 Python 中的管道操作符 |，将多个 Runnable 实例像水管一样拼接成一个 RunnableSequence（可运行序列，也就是我们常说的“链”）。

工作流转机制极度优雅：在一条链中，上一个可运行对象的 invoke() 输出，会自动作为输入传递给下一个对象，开发者根本不需要手动去提取和传递中间变量数据。

例如：chain = prompt | model | parser。这行代码就定义了一个从构建提示词、到调用模型、再到解析输出的完整流水线。

在实战之前我先说一个重要的事情，关于api-key的申请这里大家可以去查一下资料，一般在模型的光网就可以申请得到，但是重要的是一定要保证自己api-key的隐私性，**切记不可以将自己的密钥暴露在公网中，**所以在实际的开发中，我们可以将自己的api-key保存在自己的环境变量中，比如我申请了open_ai模型的api-key,我就可以在自己的电脑环境变量中设置一个名为 OPENAI_API_KEY的环境变量，这样大家在进行本地开发的时候程序会自动识别到。

三、构建基础工作流的三大支柱

让我们来看看组成上面那个基础链条（Prompt -> Model -> Parser）的三大支柱到底是什么。

1. 消息 (Messages) 与聊天模型 (Chat Models)

我们需要区分传统的“纯文本模型”和“聊天模型”。现代的框架更偏向使用聊天模型。聊天模型接受的不再是一大段纯文本，而是一个消息列表，每条消息都带有明确的角色：

SystemMessage (系统消息)：用于设定 AI 的人设和全局行为准则。
HumanMessage (人类消息)：用户的具体输入和问题。
AIMessage (AI消息)：模型返回的回答。

2. 提示词模板 (Prompt Templates)

硬编码提示词是开发大忌。Prompt Templates 的作用是将“提示词结构”与“动态注入的数据”分离开来。

通过 ChatPromptTemplate，我们可以预先定义好包含变量占位符（如 {user_input}）的系统指令和模板，极大地保障了格式的统一性和复用性。

进阶技巧：针对难以描述的复杂格式（如信息提取），我们可以使用 FewShotChatMessagePromptTemplate 为模型提供少样本示例（Few-Shot），模型往往能立刻心领神会。

3. 输出解析器 (Output Parsers)

这是解决模型“非结构化输出”痛点的关键组件。它位于链条的末端，负责将 AIMessage 转换成我们程序需要的格式。

StrOutputParser：最常用，将模型的响应直接提取为干净的纯文本字符串。
结构化解析器：通过结合 Pydantic 或 JSON 解析器，它可以强制模型输出带有指定字段的严格 JSON 结构，方便代码直接读取和操作。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 初始化模型 (支柱一)
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 2. 定义提示词模板 (支柱二)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深的翻译专家，请将用户输入的文本翻译成带有脱口秀风格的幽默中文。"),
    ("human", "{text}") # {text} 是等待注入的变量
])

# 3. 定义输出解析器 (支柱三)
parser = StrOutputParser()

# 4. 见证奇迹：用 LCEL 将它们串联成一条链！
chain = prompt | model | parser

# 5. 调用执行 (invoke)
result = chain.invoke({"text": "I was stuck in traffic for two hours today."})
print(result) 
# 输出示例：老铁，我今天在马路上硬生生当了两个小时的“车模”，那叫一个堵啊！

四、赋予大模型“双手”：工具集成 (Tools)

大模型无法直接获取实时信息或操作外部系统，这时我们需要为其装上“双手”——工具（Tools）。

1. 创建自定义工具

在 LangChain 中，通过 @tool 装饰器可以将任意 Python 函数转化为大模型可用的工具。

重点注意：在编写工具函数时，**类型提示（Type Hints）和明确的文档字符串（Docstring）**是必不可少的。因为 LangChain 会在底层自动将这些类型和注释转换为模型能看懂的 JSON Schema 描述，模型就是靠阅读这些描述来决定何时以及如何调用你的工具的。

2. 绑定工具与工具调用闭环

定义好工具后，我们使用模型的 .bind_tools() 方法将工具（比如“天气查询 API”、“计算器”）绑定到模型上。

此时，交互会形成一个闭环：

用户提问。
模型思考后认为需要调用工具，它不会直接返回答案，而是返回一个带有 tool_calls 属性的 AIMessage。
开发者（或框架）在本地执行对应的工具函数。
将工具执行的结果构造成一个特殊的 ToolMessage，并连同之前的聊天历史再次传给大模型。
大模型综合工具返回的结果，最终生成自然语言答案返回给用户。

(附注：除了自定义工具，LangChain 官方也提供了海量开箱即用的内置工具箱，如 Google Search、Wikipedia 检索等。)

from langchain_core.tools import tool

# 1. 使用 @tool 装饰器定义工具
@tool
def get_weather(city: str) -> str:
    """当用户询问天气时，调用此工具。需要传入城市名称。"""
    # 这里模拟一个 API 请求
    if "北京" in city:
        return "北京今天晴天，气温 25°C，适合穿短袖。"
    return "未知天气"

# 2. 将工具绑定到模型上
llm_with_tools = model.bind_tools([get_weather])

# 3. 测试工具调用
response = llm_with_tools.invoke("北京今天天气怎么样啊？")

# 4. 验证模型是否决定使用工具
print(response.tool_calls)
# 输出结构会明确告诉你：模型想要调用 &#x27;get_weather&#x27; 工具，并传入了参数 {&#x27;city&#x27;: &#x27;北京&#x27;}

五、赋予大模型“私有大脑”：RAG（检索增强生成）基础

解决“知识陈旧与缺乏私有知识”的最终兵器是 RAG (Retrieval-Augmented Generation) 架构。它相当于给大模型外接了一个“私有大脑”。一个标准的 RAG 包含五大流程：

文档加载 (Document Loading)：使用相应的加载器（如 UnstructuredMarkdownLoader、PDFLoader）读取企业内部的非结构化数据。
文本分割 (Splitting)：整本书模型是读不下的。需要使用文本分割器（Text Splitters）将大文档切分为包含上下文的合理小块（Chunks）。
存储 (Storage)：借助嵌入模型（Embeddings），将这些文本块转化为高维向量，并存入专用的向量数据库（Vector DB）。
检索 (Retrieval)：当用户提问时，通过相似度搜索（Similarity）或最大边际相关性（MMR）算法，从向量库中揪出与问题最相关的几个文档片段。
生成 (Generation)：最后，将“用户的原始问题”连同“检索出的文档片段”拼接到 Prompt 中，一并交给大模型，让它基于检索到的知识生成准确的答案。

利用 LCEL 构建 RAG 链极其丝滑：

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.runnables import RunnablePassthrough

# --- 第一阶段：数据准备 (线下执行) ---

# 1. 加载文档 (假设本地有个 company_policy.txt 包含公司规章制度)
# with open("company_policy.txt", "w") as f: f.write("公司规定：迟到10分钟扣50元。")
loader = TextLoader("company_policy.txt")
docs = loader.load()

# 2. 文本分割 (切分成适合大模型胃口的小块)
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
splits = text_splitter.split_documents(docs)

# 3. 向量存储 (将文本转为向量并存入 FAISS 本地内存库)
vectorstore = FAISS.from_documents(documents=splits, embedding=OpenAIEmbeddings())


# --- 第二阶段：检索与生成 (线上执行) ---

# 4. 创建检索器 (Retriever)
retriever = vectorstore.as_retriever()

# 定义 RAG 专属提示词
template = """请基于以下检索到的背景知识来回答用户问题。如果背景知识中没有相关内容，请回答“我不知道”。
背景知识：{context}

用户问题：{question}"""
rag_prompt = ChatPromptTemplate.from_template(template)

# 5. 组装终极 RAG 链！
# RunnablePassthrough() 会把用户输入的原封不动地传递给 question 变量
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | StrOutputParser()
)

# 测试问答
answer = rag_chain.invoke("请问员工迟到了怎么罚款？")
print(f"AI 回答: {answer}")
# AI 回答: 根据公司规定，迟到10分钟扣50元。

因为检索器本身也是一个 Runnable 对象，我们可以结合 RunnablePassthrough（它的作用是在链中透明传递用户的原始输入），将上述流程一气呵成连入基础链条：

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt 
    | model 
    | StrOutputParser()
)

仅仅几行代码，一个完整的私有知识问答管线就搭建完成了！

六、总结与下期预告

总结：从上面的实战我们可以看出，LangChain 的核心优势在于它极其擅长通过统一的 Runnable 接口和优雅的 LCEL 管道语法，快速编排线性、预定义的工作流（例如上面那条一气呵成的 RAG 数据管线）。

痛点过渡：然而，真实世界往往是复杂的。如果你的业务需求变成了：“让大模型自主根据当前环境动态判断下一步该干嘛、执行工具出错后自动重试、甚至多次循环推理纠错”，这种非线性的需求如果强行用 LangChain 的“链”结构和 Python 的 if-else 去实现，代码会变得极其臃肿和痛苦。

下期预告：

单线任务我们已经通关。在下一篇文章中，我们将引出这一系列的最终主角——LangGraph。我将带大家正式跳出“链（Chain）”的束缚，进入非线性、状态化、支持循环的“图（Graph）”工作流世界，手把手解锁复杂的 AI Agent（智能体）开发实战！敬请期待！

一、 引言：为什么我们需要 LangChain？

二、 核心基石：Runnable 接口与 LCEL 语法

1. 统一的 Runnable 接口

2. LangChain 表达式语言 (LCEL)

三、 构建基础工作流的三大支柱

1. 消息 (Messages) 与 聊天模型 (Chat Models)