LangChain 实操指南：从零构建生产级 LLM 应用

📋 文章导航

• 前言：为什么你需要 LangChain？
• 第一章：环境搭建与核心概念
• 第二章：Model I/O —— 与大模型对话的基础
• 第三章：检索增强生成（RAG）—— 让LLM访问你的私有数据
• 第四章：Chain —— 构建多步骤工作流
• 第五章：Agent —— 让LLM自主使用工具
• 第六章：生产部署要点（避坑指南）
• 终章：一个完整的端到端示例

前言：为什么你需要 LangChain？

2026年，大语言模型已经无处不在。但一个残酷的现实是：仅靠API调用GPT，你只能做Demo，做不了产品。 真正的价值在于——让LLM与外部工具交互、访问私有数据、执行多步骤推理。这正是LangChain登场的理由。

LangChain不是"又一个SDK"，而是一套构建LLM驱动应用的完整框架。它抽象了Prompt管理、输出解析、检索增强、Agent决策等核心环节，让你从"调API工程师"进化到"AI应用架构师"。

本文不会面面俱到地罗列API文档，而是从一个真实的企业知识库问答系统需求出发，带你走通从环境搭建到生产部署的全链路。每一段代码都经过验证，可直接运行。

第一章：环境搭建与核心概念

1.1 安装与环境准备

# 推荐使用Python 3.11+
pip install langchain langchain-openai langchain-community
如需本地模型（本文后期会用到）
pip install langchain-ollama
向量数据库（用于RAG）
pip install chromadb

设置你的OpenAI API Key（或使用国内代理）：

import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
# 如使用代理
os.environ["OPENAI_API_BASE"] = "https://your-proxy.com/v1"

1.2 LangChain 的三大核心抽象

在动手写代码前，先建立一张"认知地图"：

LangChain表达式语言（LCEL）是贯穿所有组件的"通用语"，它用 | 管道操作符将各环节串联起来，简洁且可观测。

第二章：Model I/O —— 与大模型对话的基础

2.1 统一多模型调用接口

LangChain最实用的价值之一：一套代码，切换GPT、Claude、Llama或国产模型。

from langchain_openai import ChatOpenAI
from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage, SystemMessage
云端模型
gpt = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
本地模型（需先运行 ollama pull llama3.2）
llama = ChatOllama(model="llama3.2", temperature=0)
统一的消息格式
messages = [
SystemMessage(content="你是一位精通Python的性能优化专家"),
HumanMessage(content="如何快速判断一个列表是否包含重复元素？给出三种方法并比较时间复杂度")
]
response = gpt.invoke(messages)
print(response.content)

关键点：invoke() 是LangChain所有组件的统一入口，未来还会用到 stream()（流式输出）和 batch()（批量处理）。

2.2 PromptTemplate —— 告别字符串拼接

from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "你是一位{role}，擅长用{style}风格解释复杂概念"),
("human", "请用{word_count}个字以内解释：{topic}")
])
chain = prompt | gpt
result = chain.invoke({
"role": "物理老师",
"style": "幽默风趣",
"word_count": 50,
"topic": "量子纠缠"
})
print(result.content)

2.3 输出解析器 —— 让AI输出JSON/结构化数据

很多初学者会踩的坑：让AI输出JSON，结果它返回了Markdown代码块。用输出解析器解决：

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
定义你期望的强类型结构
class CodeReview(BaseModel):
bug_risk: int = Field(description="潜在Bug风险评分，1-10")
suggestions: list[str] = Field(description="改进建议列表")
refactored_code: str = Field(description="重构后的代码片段")
parser = PydanticOutputParser(pydantic_object=CodeReview)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个严格的代码评审专家。{format_instructions}"),
("human", "请评审以下Python代码：\\npython\\n{code}\\n")
]).partial(format_instructions=parser.get_format_instructions())
chain = prompt | gpt | parser # 自动解析为CodeReview对象
review = chain.invoke({"code": "def add(a,b): return a+b"})
print(f"风险评分: {review.bug_risk}")
print(f"建议: {review.suggestions[0]}")

第三章：检索增强生成（RAG）—— 让LLM访问你的私有数据

这是LangChain最广泛的应用场景。我们以公司内部文档问答为例。

3.1 数据加载与切分

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
加载指定目录下的所有txt文件
loader = DirectoryLoader("./docs/", glob="**/*.txt", loader_cls=TextLoader)
docs = loader.load()
智能切分：按段落、句子递归切割，保留语义完整性
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50, # 重叠部分防止语义断层
separators=["\\n\\n", "\\n", "。", "！", "？", " ", ""]
)
chunks = splitter.split_documents(docs)
print(f"原始文档{len(docs)}份，切分为{len(chunks)}个片段")

3.2 向量化与存储

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
生成向量嵌入（也可用本地模型如BGE）
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
存入向量数据库
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
持久化保存，下次直接加载
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

3.3 构建RAG链 —— 检索 + 生成

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # 返回最相似的4个片段
模板：让LLM基于检索到的上下文回答
template = """你是一位专业助手。请基于以下【上下文】回答用户问题。
如果上下文不足以回答问题，请直接说"根据现有文档无法回答该问题"。
【上下文】
{context}
【用户问题】
{question}
请给出准确、简洁的回答："""
prompt = ChatPromptTemplate.from_template(template)
LCEL方式构建RAG链
rag_chain = (
{
"context": retriever | (lambda docs: "\\n\\n".join(d.page_content for d in docs)),
"question": RunnablePassthrough()
}
| prompt
| gpt
)
执行查询
question = "公司的年假制度是怎样的？"
response = rag_chain.invoke(question)
print(response.content)

这一小段代码就是RAG的核心，不到20行实现了"检索→拼接→生成"的完整链路。

第四章：Chain —— 构建多步骤工作流

4.1 简单的顺序链

from langchain_core.runnables import RunnableLambda
步骤1：提取关键词
extract_prompt = ChatPromptTemplate.from_template("提取以下文本中的3个核心关键词，用逗号分隔：\\n{text}")
extract_chain = extract_prompt | gpt
步骤2：基于关键词生成摘要
summary_prompt = ChatPromptTemplate.from_template("基于关键词[{keywords}]，为以下文本写一段100字摘要：\\n{text}")
summary_chain = summary_prompt | gpt
组合成顺序链（注意：这里用RunnablePassthrough传递原始text）
full_chain = (
RunnablePassthrough.assign(keywords=extract_chain)
| summary_chain
)
result = full_chain.invoke({"text": "（你的长文本内容）"})

4.2 分支路由链（Router Chain）

# 根据问题类型路由到不同的处理链
router_template = """根据问题内容，分类为以下之一：['技术', '财务', '人事']。
只输出分类名称。
问题：{question}"""
router_chain = ChatPromptTemplate.from_template(router_template) | gpt
def route_output(output):
category = output.content.strip()
if category == "技术":
return tech_chain # 假设已定义
elif category == "财务":
return finance_chain
else:
return hr_chain
使用RunnableBranch实现条件路由（略，思路如上）

第五章：Agent —— 让LLM自主使用工具

Agent是LangChain最"智能"的部分。LLM不再是"一次性回答"，而是循环思考→决定调用工具→观察结果→继续思考，直至完成任务。

5.1 定义工具（Tools）

from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain_core.tools import tool
内置工具
search = DuckDuckGoSearchRun()
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
自定义工具（使用@tool装饰器）
@tool
def calculate(expression: str) -> str:
"""安全地计算数学表达式，如'2+3*4'"""
try:
return str(eval(expression))
except:
return "表达式无效"
@tool
def get_current_time() -> str:
"""获取当前日期和时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
tools = [search, wikipedia, calculate, get_current_time]

5.2 创建并运行Agent

from langchain.agents import create_tool_calling_agent, AgentExecutor
使用OpenAI的工具调用能力（推荐）
agent = create_tool_calling_agent(gpt, tools, ChatPromptTemplate.from_messages([
("system", "你是一个乐于助人的助手，可以调用工具获取信息。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
]))
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
执行复杂任务
result = agent_executor.invoke({
"input": "现在几点了？然后帮我查一下明天北京天气，最后计算一下3.14*5的平方"
})
print(result["output"])

Agent会依次调用 get_current_time → 搜索天气 → calculate，并在每一步观察结果后决定下一步。

第六章：生产部署要点（避坑指南）

6.1 回调与可观测性

from langchain_core.callbacks import StdOutCallbackHandler
调试时打印详细日志
chain.invoke(input_data, config={"callbacks": [StdOutCallbackHandler()]})
生产环境建议接入LangSmith（LangChain官方监控平台）
os.environ["LANGCHAIN_API_KEY"] = "your-key"
chain.invoke(input_data, config={"run_name": "my-rag-chain"})

6.2 异步与流式处理

# 异步调用
response = await chain.ainvoke(input_data)
流式输出（用户体验更好）
async for chunk in chain.astream(input_data):
print(chunk.content, end="", flush=True)
批量处理（高吞吐场景）
results = chain.batch([{"question": "q1"}, {"question": "q2"}])

6.3 缓存与成本控制

from langchain_core.caches import InMemoryCache
from langchain.globals import set_llm_cache
set_llm_cache(InMemoryCache()) # 相同输入直接返回缓存结果
生产环境可用Redis缓存：langchain_community.cache.RedisCache

6.4 常见陷阱