LangChain 1.x 工程化实践：从 LLM 调用到智能体与 RAG 应用开发
1. 从“玩具”到“工程”：为什么我们需要 LangChain
如果你在过去一年里尝试过用大语言模型（LLM）来构建点什么东西，大概率经历过这样一个过程：一开始，你兴奋地在 OpenAI 的 Playground 里输入几个问题，模型给出的回答让你惊叹不已。然后你开始琢磨，能不能让它读取我自己的文档？能不能让它记住我们之前的对话？能不能让它调用一些外部工具，比如查查天气或者发封邮件？
于是你打开 API 文档，开始写代码。你很快发现，要让 LLM 完成一个稍微复杂点的任务，你需要处理一堆琐碎但关键的事情：如何把用户的问题、历史对话、相关文档片段拼成一个符合模型要求的提示词（Prompt）？如何把模型输出的文本，解析成结构化的数据，比如 JSON 对象，好让程序能接着处理？当任务需要多步推理或调用多个工具时，如何设计一个清晰的流程来控制这一切？
最初，你可能用一堆if-else和字符串拼接硬写出来一个能跑的原型。但很快，代码就变得难以维护和扩展。你意识到，你正在重复发明轮子，而且这个轮子还不太好用。这时，你听说了 LangChain。
LangChain 本质上是一个框架，它把使用 LLM 构建应用时那些通用、繁琐但又至关重要的部分，抽象成了可复用的组件和设计模式。它不是一个“魔法黑盒”，而更像是一套精心设计的“乐高积木”和“搭建说明书”。在 1.x 版本中，这套积木经过了大规模的重构和简化，目标就是让开发者能更直观、更高效地搭建出稳定、可维护的 AI 应用。简单说，LangChain 帮你把 LLM 从一个有趣的“玩具”，变成了可以嵌入真实业务流的“工程化组件”。
2. LangChain 1.x 的核心设计哲学：清晰、简洁、可组合
LangChain 早期版本（如 0.0.x 系列）功能强大，但概念较多，学习曲线陡峭。1.x 版本是一次重大的理念升级，其核心设计哲学可以概括为三点，这直接决定了我们如何使用它。
2.1 以“链”（Chain）为中心，但链更轻量
在 LangChain 中，“链”是将多个组件（模型、提示词、输出解析器、工具等）串联起来执行特定任务的蓝图。1.x 版本强化了“链”作为首要抽象的概念，但让链的构建变得更加声明式和直观。
过去你可能需要继承一个基类并实现_call方法。现在，更推荐使用LCEL（LangChain Expression Language）。LCEL 是一种声明式的语法，让你能用|操作符像连接管道一样连接各个组件，代码清晰且支持流式输出、异步等特性。
# 一个简单的 LCEL 链示例：生成公司名 -> 生成口号 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4") prompt_for_name = ChatPromptTemplate.from_template("为生产{product}的公司起个名字，只返回名字。") prompt_for_slogan = ChatPromptTemplate.from_template("为这家名为{company_name}的公司写一句口号。") chain = prompt_for_name | model | (lambda x: {"company_name": x.content}) | prompt_for_slogan | model # 执行链 result = chain.invoke({"product": "环保咖啡杯"}) print(result.content)这个例子中，|清晰地展示了数据流：产品信息经过第一个提示词模板，交给模型生成公司名，然后提取出公司名文本，再送入第二个提示词模板，最后交给模型生成口号。整个逻辑一目了然。
2.2 明确的模块化与“命名空间”分离
LangChain 1.x 将庞大的库拆分成多个独立、专注的包，形成了清晰的“命名空间”。这带来了几个好处：
- 依赖更干净：你的项目只需要引入真正用到的包，减少了依赖冲突和臃肿。
- 概念更清晰：每个包的职责边界明确。
- 升级更稳定：核心接口 (
langchain-core) 保持稳定，其他包可以独立迭代。
主要包包括：
langchain-core： 核心抽象和运行时。BaseMessage,Runnable接口，LCEL都在这里。这是构建任何链的基础，几乎必须引入。langchain： “元包”，通常通过pip install langchain安装，它本身不包含太多代码，主要作用是拉取一系列常用的集成包（如langchain-openai,langchain-community），方便初学者快速开始。对于生产项目，更推荐直接安装你需要的特定包。langchain-community： 社区维护的第三方集成，比如一些不太常用的模型封装、工具或向量数据库接口。稳定性可能略低于官方维护的包。langchain-<provider>： 官方维护的特定供应商集成包。例如langchain-openai,langchain-anthropic,langchain-mistralai。这些包提供了对该供应商模型最稳定、最新的支持。langchain-<use-case>： 针对特定用例的高级包，如langchain-text-splitters（文本分割），langchain-chroma（Chroma 向量库集成）。
实操心得： 在新项目中，我通常会直接从
langchain-core和具体的供应商包（如langchain-openai）开始安装。只有当需要用到社区工具或特定向量库时，才引入langchain-community或相应的集成包。这能最大程度保持项目依赖的简洁和可维护性。
2.3 “一切皆可运行”（Everything is Runnable）
这是 1.x 版本一个非常强大的统一抽象。在langchain-core中，定义了Runnable协议。一个Runnable对象可以是：
- 一个提示词模板 (
PromptTemplate) - 一个大语言模型 (
ChatOpenAI) - 一个工具 (
Tool) - 一个输出解析器 (
StrOutputParser,JsonOutputParser) - 甚至是你自己写的一个函数（通过
RunnableLambda包装）
最关键的是，任何实现了Runnable接口的对象，都可以用相同的方式调用（invoke,batch,stream），并且可以用|连接起来组成链。这种一致性极大地简化了开发和调试。你不再需要记住某个组件是调用run还是predict或是invoke，统一使用invoke即可。
from langchain_core.runnables import RunnableLambda def double_length(text: str) -> dict: return {"length_doubled": len(text) * 2} runnable_func = RunnableLambda(double_length) # 可以轻松地将其接入链中 chain = model | runnable_func result = chain.invoke("Hello world") print(result) # 输出: {'length_doubled': 22}3. 核心组件深度拆解：不只是调用模型
理解 LangChain，必须超越“它帮我调了 API”这个层面。它的价值在于提供了一套处理 LLM 输入输出的标准化“流水线”。我们拆解这条流水线上的几个关键工位。
3.1 提示词管理：从字符串模板到结构化消息
直接拼接字符串构造提示词是万恶之源，难以维护且易出错。LangChain 的PromptTemplate和ChatPromptTemplate提供了解决方案。
ChatPromptTemplate是更现代、更推荐的方式，因为它直接对应底层聊天模型（如 GPT-4）的消息格式。它允许你构建一个由SystemMessage、HumanMessage、AIMessage等组成的消息列表。
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate system_template = SystemMessagePromptTemplate.from_template( "你是一位专业的{domain}专家，回答问题时请务必严谨，并引用相关知识。" ) human_template = HumanMessagePromptTemplate.from_template("请解释一下{concept}。") chat_prompt = ChatPromptTemplate.from_messages([system_template, human_template]) # 格式化后，得到的是一个 `List[BaseMessage]`，直接可以喂给聊天模型 formatted_messages = chat_prompt.format_messages(domain="量子物理", concept="量子纠缠") print(formatted_messages) # [SystemMessage(content='你是一位专业的量子物理专家...'), # HumanMessage(content='请解释一下量子纠缠。')]为什么这很重要？因为复杂的应用往往需要动态组合提示词。比如在检索增强生成（RAG）中，你需要把检索到的文档片段插入到提示词的特定位置。使用ChatPromptTemplate，你可以轻松地通过+操作符组合不同的提示部分，或者使用partial方法预先填充部分变量，这让提示词工程变得模块化和可测试。
3.2 输出解析：让非结构化的文本“就范”
LLM 输出的是非结构化的文本。但我们的程序需要结构化的数据。输出解析器（OutputParser）就是负责把文本转换成我们需要格式的组件。
StrOutputParser: 最简单的解析器，就是提取模型的文本输出。JsonOutputParser: 要求模型输出 JSON，并自动将其解析为 Python 字典或列表。你需要提供一个 JSON Schema 或 Pydantic 模型来指导模型。
from langchain_core.output_parsers import JsonOutputParser from langchain_core.pydantic_v1 import BaseModel, Field class Joke(BaseModel): setup: str = Field(description="笑话的开头部分") punchline: str = Field(description="笑话的包袱或结尾") parser = JsonOutputParser(pydantic_object=Joke) # 在提示词中，我们可以通过 `parser.get_format_instructions()` 获取指导模型输出 JSON 的指令 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个讲笑话的助手。{format_instructions}"), ("human", "讲一个关于{subject}的笑话。") ]).partial(format_instructions=parser.get_format_instructions()) chain = prompt | model | parser result = chain.invoke({"subject": "程序员"}) print(result) # 输出: Joke(setup='为什么程序员分不清万圣节和圣诞节？', punchline='因为 Oct 31 == Dec 25！') # 此时 result 是一个 Joke 对象，可以直接访问 result.setup 和 result.punchline- 自定义解析器： 通过继承
BaseOutputParser或使用RunnableLambda，你可以处理任何复杂的解析逻辑，比如从一段文本中提取特定关键词、解析成特定的数据结构等。
踩坑实录： 使用
JsonOutputParser时，最大的坑在于模型有时会“自言自语”，在 JSON 前后添加额外的解释性文字，导致解析失败。解决方案是：第一，在系统提示词中明确强调“只输出 JSON，不要有任何额外文本”；第二，使用更强大的模型（如 GPT-4）；第三，在自定义解析器中加入后处理逻辑，尝试用正则表达式从响应中提取 JSON 部分。LangChain 内置的解析器通常已经具备一定的容错能力。
3.3 记忆机制：让对话拥有“上下文”
无状态的 HTTP 请求如何让 LLM 记住之前说过的话？这就是记忆（Memory）组件的作用。LangChain 提供了多种记忆后端。
- 对话缓存记忆 (
ConversationBufferMemory)： 最简单，将整个对话历史以字符串形式保存在内存中。缺点是对话长了之后，会消耗大量 Token，且可能超出模型上下文长度。 - 对话摘要记忆 (
ConversationSummaryMemory)： 每次交互后，用一个单独的 LLM 调用去总结之前的对话历史，只保留摘要。这能显著缩短历史长度，但会丢失细节，且增加了成本和延迟。 - 向量存储记忆 (
ConversationVectorStoreMemory)： 将每次对话的消息嵌入成向量，存入向量数据库（如 Chroma）。当需要回忆时，根据当前问题检索最相关的历史片段。这种方式能高效利用长上下文，并且回忆的内容更相关，是构建复杂长期记忆系统的方向。
关键是如何将 Memory 集成到链中？在 1.x 的 LCEL 范式下，推荐使用RunnableWithMessageHistory。它包装你的链，并自动处理历史消息的注入和保存。
from langchain.memory import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory # 1. 定义你的核心链（无记忆） prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个友好的助手。"), MessagesPlaceholder(variable_name="history"), # 占位符，用于插入历史消息 ("human", "{input}") ]) chain = prompt | model # 2. 定义一个存储工厂，用于获取或创建每个会话的历史记录 store = {} def get_session_history(session_id: str) -> ChatMessageHistory: if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] # 3. 用 RunnableWithMessageHistory 包装链 chain_with_memory = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="input", # 用户当前输入对应的变量名 history_messages_key="history", # 提示词中历史消息占位符的变量名 ) # 4. 调用时传入 session_id response = chain_with_memory.invoke( {"input": "我叫小明。"}, config={"configurable": {"session_id": "user_123"}} ) response = chain_with_memory.invoke( {"input": "我刚才说我叫什么？"}, config={"configurable": {"session_id": "user_123"}} # 相同的 session_id 会取出历史 )这种方式将记忆管理与业务逻辑链解耦，非常清晰。
4. 高级模式：智能体与检索链的工程化实现
掌握了基础组件，我们就可以搭建更复杂的应用。LangChain 最常被用于两种模式：检索增强生成（RAG）和智能体（Agent）。
4.1 构建生产级 RAG 链：超越简单的问答
一个基础的 RAG 链是：用户提问 -> 检索相关文档 -> 将文档和问题组合成提示词 -> 模型生成答案。但生产环境要求更高。
挑战一：检索质量。简单的向量相似度搜索可能返回不相关或冗余的内容。LangChain 提供了多种检索器（Retriever）和检索后处理技术：
MultiQueryRetriever： 让 LLM 基于原始问题生成多个相关问题，并行检索，合并去重后返回结果。这能提高召回率。ContextualCompressionRetriever： 在检索后，使用一个单独的 LLM 调用对检索到的文档进行压缩、过滤或总结，只保留与问题最相关的部分再送入最终提示词。这能提升精度并节省 Token。EnsembleRetriever： 结合不同检索方式（如向量检索和关键词检索）的结果，取长补短。
挑战二：答案的忠实性与引用。模型可能“幻觉”出不存在于文档中的信息。我们需要让答案可追溯。
- 在提示词中明确要求模型基于给定上下文回答，并注明“如果上下文未提供相关信息，请回答‘我不知道’”。
- 使用
ChatPromptTemplate的MessagesPlaceholder动态插入检索到的文档。 - 在输出解析环节，可以设计让模型同时输出答案和引用的文档 ID 或片段。
from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # 假设已有向量库 `vectorstore` embeddings = OpenAIEmbeddings() vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) base_retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) # 使用 LLM 进行上下文压缩 compressor = LLMChainExtractor.from_llm(model) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=base_retriever ) # 构建 RAG 链 template = """基于以下上下文回答问题。如果你不知道答案，就说你不知道。 上下文：{context} 问题：{question} 请提供详细的答案，并注明答案来源于上下文的哪些部分（如果有的话）。""" prompt = ChatPromptTemplate.from_template(template) rag_chain = ( {"context": compression_retriever, "question": RunnablePassthrough()} | prompt | model | StrOutputParser() )4.2 智能体：让 LLM 学会使用工具
智能体是 LangChain 的另一个高光特性。其核心思想是：LLM 作为“大脑”，根据用户目标，自主决定是否调用、以及按什么顺序调用哪些“工具”（函数），并解析工具返回的结果，最终完成任务。
工具（Tool）： 任何可以被 LLM 调用的函数。需要定义清晰的名称、描述和参数。LangChain 内置了大量工具（如搜索、计算器、终端），你也可以轻松自定义。
from langchain.agents import tool from datetime import datetime @tool def get_current_time(timezone: str = "Asia/Shanghai") -> str: """获取指定时区的当前时间。""" # 这里简化实现，实际应用中应使用 pytz 等库 now = datetime.now() return f"The current time in {timezone} is approximately {now.strftime('%Y-%m-%d %H:%M:%S')}." # 创建智能体 from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # 从 LangChain Hub 拉取一个预设的 ReAct 提示词 prompt = hub.pull("hwchase17/react") # 创建智能体 agent = create_react_agent(model, [get_current_time], prompt) agent_executor = AgentExecutor(agent=agent, tools=[get_current_time], verbose=True) # 执行 result = agent_executor.invoke({"input": "现在上海是几点钟？"})智能体的工作流： 以上面的create_react_agent（基于 ReAct 框架）为例：
- LLM 收到用户输入和提示词（提示词中包含了工具描述和 ReAct 格式要求）。
- LLM 思考（Thought）：分析当前情况，决定下一步行动。
- LLM 行动（Action）：输出一个格式化的动作，如
Action: get_current_time，Action Input: {"timezone": "Asia/Shanghai"}。 - 框架解析这个动作，调用对应的工具函数。
- 工具返回结果（Observation），如
Observation: The current time in Asia/Shanghai is 2024-05-27 14:30:00。 - 这个观察结果被反馈给 LLM，LLM 继续思考，直到它认为可以给出最终答案（Final Answer）。
智能体执行器（AgentExecutor）负责管理这个循环：调用智能体，解析输出，执行工具，处理错误（如工具调用失败、解析失败），并控制最大迭代次数以防无限循环。
核心避坑点： 智能体非常强大，但也容易失控。关键点在于：第一，工具的描述必须极其精确。LLM 完全依赖描述来决定是否以及如何使用工具。模糊的描述会导致错误的调用。第二，设置合理的
max_iterations（如 10），避免在复杂任务中陷入死循环。第三，在生产环境中，务必对工具调用进行沙盒化和权限控制，尤其是涉及文件操作、网络请求或系统命令的工具，防止智能体执行危险操作。
5. 部署与生产化考量：从原型到服务
用 LangChain 快速搭出一个原型后，如何将它变成一个可靠的服务？
5.1 配置管理与密钥安全
永远不要将 API 密钥硬编码在代码中。LangChain 1.x 鼓励使用langchain-cli或环境变量进行管理。
# 在 .env 文件中 OPENAI_API_KEY=sk-... ANTHROPIC_API_KEY=...在代码中，通过os.getenv读取，或使用langchain的配置加载功能。对于更复杂的配置（如不同环境的不同模型），可以考虑使用 Pydantic Settings 管理。
5.2 可观测性与调试
当链变得复杂时，调试输出变得困难。LangChain 提供了回调系统（Callbacks），允许你在链执行的各个阶段（如on_chain_start,on_llm_end）插入钩子函数，用于记录日志、追踪性能或流式传输中间结果。
from langchain.callbacks.stdout import StdOutCallbackHandler chain = prompt | model result = chain.invoke({"topic": "AI"}, config={"callbacks": [StdOutCallbackHandler()]})StdOutCallbackHandler会在控制台打印出详细的执行步骤和耗时，是调试的利器。在生产中，你可以实现自定义的 CallbackHandler，将日志发送到如 LangSmith、Prometheus 等监控系统。
5.3 性能优化与成本控制
- 缓存： 对模型调用进行缓存可以大幅减少重复请求，降低成本和延迟。LangChain 支持内存缓存 (
InMemoryCache)、SQLite 缓存、Redis 缓存等。对于提示词固定、仅输入参数变化的场景（如翻译不同句子），缓存效果极佳。 - 批处理： 使用链的
batch方法一次性处理多个输入，某些云服务商（如 OpenAI）的批处理 API 有更低的价格。 - 流式输出： 对于需要长时间生成内容的场景，使用
stream方法可以边生成边返回，提升用户体验。LCEL 链天然支持流式。 - 模型选型： 并非所有任务都需要 GPT-4。对于简单的分类、提取任务，
gpt-3.5-turbo甚至更小的开源模型可能就足够了。在链的不同环节使用不同性价比的模型，是控制成本的有效策略。
5.4 测试与评估
AI 应用的非确定性输出使得测试变得挑战。LangChain 社区和 LangSmith 提供了评估框架。你可以定义评估函数（例如，检查输出是否包含特定关键词、是否与参考答案语义相似），然后在一组测试用例上运行你的链，自动计算通过率。这是确保应用质量迭代的关键。
6. 从 0.x 到 1.x：迁移指南与心态调整
如果你有基于 LangChain 0.x 的老项目，升级到 1.x 可能需要一些工作，但收益是长期的代码清晰度和可维护性。
主要变化与迁移步骤：
- 包结构： 将
from langchain.llms import OpenAI改为from langchain_openai import OpenAI（或ChatOpenAI）。类似地，检查所有导入，使用新的独立包。 - 链的构建： 这是最大的变化。将旧的
LLMChain、SequentialChain等，重写为基于 LCEL 的声明式链。虽然需要重写，但新链更简洁、功能更强（如原生支持流式、异步）。 - 方法调用： 统一使用
invoke（单次调用）、batch（批量）、stream（流式）和ainvoke、abatch（异步版本）。替换旧的run、predict、apply等方法。 - 记忆集成： 如前所述，改用
RunnableWithMessageHistory模式，替代旧的在链构造函数中传入memory参数的方式。 - 智能体： 智能体的创建 API 在 1.x 也有更新，建议查阅最新文档，使用新的
create_react_agent、create_openai_tools_agent等工厂函数。
心态调整： 不要将 LangChain 1.x 视为一个简单的库更新，而是视为一次框架理念的升级。它要求开发者更清晰地思考数据流（LCEL 的|操作符完美体现了这一点），更明确地管理依赖（独立的包），并拥抱更一致、更强大的抽象（Runnable）。初期学习成本可能不降反升，但一旦掌握，构建和维护复杂 AI 应用的效率会大幅提升。
LangChain 1.x 不再仅仅是一个帮你调用 API 的封装，它是一套用于编排 LLM 工作流的、具有良好设计模式的框架。它的价值在于，当你的应用从“快速验证想法”的原型阶段，迈向“稳定、可扩展、易维护”的生产阶段时，它能提供那条虽然需要学习但绝对值得的“工程化路径”。