我花了两周啃完LangChain，踩的坑比写的代码还多

最近被AI应用开发搞得有点上头。事情是这样的，公司领导突然说要搞个"智能知识库问答系统"，让内部的运维文档、故障处理手册能通过对话的方式查询。需求倒是不复杂，但我一个搞运维的，平时写写shell脚本、搞搞自动化部署还行，这突然让我去搞AI应用开发，属实有点赶鸭子上架了。

调研了一圈之后，LangChain这个框架出现频率最高。GitHub上13万多star，各种教程满天飞，看着好像挺成熟的。于是我花了大概两周时间，从零开始学了一遍，中间踩了不少坑，也算是把核心的东西搞明白了。今天把这个过程记录下来，给同样想入门的兄弟们一个参考。

项目地址：https://github.com/langchain-ai/langchain

LangChain到底解决了什么问题

在说LangChain之前，我得先说说为什么需要它。

你想啊，现在的大模型（GPT-5、Claude4、通义千问这些）确实很强，但它们本质上就是一个"输入文本→输出文本"的黑盒子。你要做一个真正能用的AI应用，光有模型是远远不够的。

举个具体的例子。我想做的知识库问答系统，需要完成这些事情：

把公司几百份运维文档（PDF、Markdown、Word都有）读进来
把文档切成小块存到向量数据库里
用户提问的时候，先去向量数据库里找相关的文档片段
把找到的片段和用户的问题一起发给大模型
大模型根据这些上下文生成回答

如果没有框架，这些东西全得自己从头写。文档解析要写、文本分割要写、向量存储要写、Prompt拼接要写……你写个demo可能还行，但要做成生产可用的东西，工作量大得吓人。

LangChain就是把这些东西都封装好了，你像搭积木一样把各个组件拼起来就行。它提供了统一的接口来对接各种大模型、各种向量数据库、各种文档格式，省了巨多重复劳动。

环境搭建

这部分我直接贴命令了，没什么好说的：

pip install langchain langchain-core langchain-community langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple/

# 向量数据库，我选的FAISS，Facebook开源的，本地跑不需要额外服务
pip install faiss-cpu -i https://pypi.tuna.tsinghua.edu.cn/simple/

# 文档加载相关的
pip install pypdf unstructured python-docx -i https://pypi.tuna.tsinghua.edu.cn/simple/

API Key的配置，我建议用.env文件别硬编码：

# .env
OPENAI_API_KEY=sk-xxxxxxxx
OPENAI_API_BASE=https://api.openai.com/v1

from dotenv import load_dotenv
load_dotenv()

这里有个坑要提一下。如果你用的是国内的代理API（比如某些转发服务），OPENAI_API_BASE这个环境变量一定要设对。我一开始设错了，报错信息还特别迷惑，排查了好久才发现是base_url的问题。

从最简单的开始：调用大模型

LangChain里调用模型非常简单：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,
    max_tokens=2000
)

response = llm.invoke("解释一下Linux的iptables和nftables有什么区别")
print(response.content)

这个没什么好说的，就是最基础的调用。但问题来了——实际场景中你不可能每次都手动拼接问题字符串对吧？所以LangChain搞了个Prompt模板的机制。

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深的{role}，回答问题要结合实际生产经验，给出可操作的建议。"),
    ("human", "{question}")
])

# 这样用
messages = prompt.format_messages(
    role="Linux运维工程师",
    question="服务器突然SSH连不上了，怎么排查？"
)
response = llm.invoke(messages)

看着好像就是个字符串格式化对吧？但实际上Prompt模板的意义在于复用和标准化。你可以把写好的模板保存下来，不同场景换个参数就能用。而且后面做Chain的时候，模板是可以直接当组件来拼接的。

LCEL：管道符拼接组件

这个是LangChain的核心语法，叫LCEL（LangChain Expression Language）。说白了就是用|把各个组件串起来：

from langchain_core.output_parsers import StrOutputParser

chain = prompt | llm | StrOutputParser()

# 一行搞定
result = chain.invoke({
    "role": "运维工程师",
    "question": "Nginx出现502错误一般怎么排查？"
})
print(result)  # 直接就是字符串了

StrOutputParser()是把模型的输出对象转成纯字符串。不加的话返回的是一个AIMessage对象，带了一堆元数据，一般用不着。

这个管道符语法刚看到的时候觉得还挺酷的，但用多了会发现有些场景不太直观。比如你想在中间加个判断逻辑，或者需要把前一步的部分输出传给后一步，写起来就有点绕了。不过对于线性流程，确实比一步步调用要简洁很多。

我后来做的一个比较实用的小东西：把运维告警信息扔给AI分析，让它判断严重程度和建议处理方案：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class AlertAnalysis(BaseModel):
    severity: str = Field(description="严重程度：critical/warning/info")
    root_cause: str = Field(description="可能的根因分析")
    suggestion: str = Field(description="处理建议")
    need_escalation: bool = Field(description="是否需要升级处理")

parser = JsonOutputParser(pydantic_object=AlertAnalysis)

alert_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个运维告警分析助手。根据告警信息进行分析，返回JSON格式的结果。
{format_instructions}"""),
    ("human", "告警内容：{alert_message}")
])

alert_chain = alert_prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0) | parser

# 测试
result = alert_chain.invoke({
    "alert_message": "[CRITICAL] 服务器web-prod-03 磁盘使用率达到95%，/var/log分区剩余空间不足2GB",
    "format_instructions": parser.get_format_instructions()
})

print(result)
# {'severity': 'critical', 'root_cause': '日志文件未及时清理或轮转配置不当...', 
#  'suggestion': '1. 先清理旧日志释放空间 2. 检查logrotate配置...', 
#  'need_escalation': False}

这个JsonOutputParser是个好东西，可以让大模型的输出直接变成结构化的Python字典。配合Pydantic定义schema，输出格式非常稳定。但也不是百分百靠谱，偶尔模型还是会输出不合规的JSON，生产环境里得加个try-except兜底。

重头戏：RAG实现

RAG（Retrieval-Augmented Generation）是我学LangChain最主要的目的。整个流程大概是这样的：

加载文档 → 切分文本 → 向量化存储 → 检索 → 生成回答

一步步来。

加载文档

from langchain_community.document_loaders import (
    PyPDFLoader,
    DirectoryLoader,
    TextLoader,
    UnstructuredMarkdownLoader
)

# 加载单个PDF
loader = PyPDFLoader("./docs/nginx运维手册.pdf")
docs = loader.load()
print(f"加载了{len(docs)}页")

# 批量加载目录下所有Markdown文件
loader = DirectoryLoader(
    "./docs/",
    glob="**/*.md",
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"},
    show_progress=True
)
docs = loader.load()

这里有个坑——TextLoader默认编码不是UTF-8，如果你的文档里有中文，不指定encoding就会报错。我在这上面浪费了至少半小时才发现。

另外DirectoryLoader加载大量文件的时候速度不快，几百个文件可能要跑个几十秒。如果文件特别多，可以考虑加个缓存机制，第一次加载完把处理好的数据存下来，后面就不用重新加载了。

文本分割

这一步非常关键，直接影响后面检索的效果：

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=80,
    separators=["\n\n", "\n", "。", "！", "？", "；", " "],
    length_function=len
)

splits = text_splitter.split_documents(docs)
print(f"分割成了{len(splits)}个片段")

chunk_size这个参数我试了好几个值。设太大（比如1000以上），检索的时候会返回一大段不太相关的内容；设太小（比如200以下），上下文信息又不够完整。对于中文技术文档，我发现400-600之间效果比较好。

chunk_overlap是重叠部分，防止在切分的时候把一句完整的话切断了。设个50-100基本够用。

separators列表的顺序是有优先级的。它会优先尝试按\n\n（段落）来切，切不下就按\n（换行），再切不下就按句号切。对中文文档来说，记得把中文标点加上去。

向量化和存储

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings()

# 构建向量数据库
vectorstore = FAISS.from_documents(splits, embeddings)

# 保存到本地，下次直接加载就不用重新计算embedding了
vectorstore.save_local("./faiss_index")

这步需要注意的是，embedding计算是要调API的，要花钱。如果你的文档量比较大（几千个片段以上），这个费用就不能忽略了。OpenAI的embedding模型相对便宜，text-embedding-3-small目前是$0.02/1M tokens，但架不住量大。

还有个经验：构建完向量数据库一定要save_local保存下来。不然每次启动程序都要重新计算，一是慢二是费钱。

如果你不想用OpenAI的embedding（比如担心数据安全问题），也可以用本地模型：

from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",
    model_kwargs={'device': 'cpu'}
)

bge-small-zh-v1.5是智源开源的中文embedding模型，效果还不错，关键是完全本地运行，不用担心数据外泄。

检索和生成

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}  # 返回最相关的4个片段
)

# 构建RAG的prompt
rag_prompt = ChatPromptTemplate.from_template("""
根据以下参考资料回答用户的问题。如果参考资料中没有相关信息，请明确告知用户你不确定，不要编造答案。

参考资料：
{context}

用户问题：{question}

请给出详细的回答：
""")

def format_docs(docs):
    return "\n\n---\n\n".join([doc.page_content for doc in docs])

# 组装RAG链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

# 提问
answer = rag_chain.invoke("Nginx配置反向代理后499错误怎么处理？")
print(answer)

这段代码里有个写法可能第一次看会有点迷糊：

{"context": retriever | format_docs, "question": RunnablePassthrough()}

它的意思是：question字段直接透传用户的输入，context字段则是先通过retriever检索文档，再用format_docs格式化成文本。两个字段并行处理，最后一起传给后面的prompt模板。

说实话这个语法我适应了一会儿才看顺眼。但理解之后确实很简洁。

RAG效果调优

RAG跑起来之后，你大概率会遇到"检索不准"或者"回答不对"的问题。我的经验是从这几个方向调：

检索阶段：

search_kwargs里的k值，太小可能漏掉重要信息，太大会引入噪音
chunk_size调整，找到最适合你文档特点的大小
试试search_type="mmr"（最大边际相关性），它会在相关性和多样性之间做个平衡，避免返回的几个片段内容太重复

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 10}  # 先取10个再从中选4个最多样的
)

生成阶段：

Prompt写好点，明确告诉模型"不知道就说不知道"
temperature设低一点（0或0.1），让输出更稳定
有时候把检索到的文档来源信息也附上，方便溯源

Agent：让AI自己决定调什么工具

这个是LangChain里我觉得最有意思的部分。简单说就是你给AI定义一些"工具"（函数），然后让它根据用户的问题自己判断该调用哪个工具。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

@tool
def check_disk_usage(hostname: str) -> str:
    """检查指定服务器的磁盘使用情况"""
    # 实际环境里这里会SSH过去执行df命令
    # demo先模拟一下
    import random
    usage = random.randint(30, 95)
    return f"服务器 {hostname} 磁盘使用率: {usage}%，/var/log 占用最大"

@tool
def check_service_status(service_name: str, hostname: str) -> str:
    """检查指定服务器上某个服务的运行状态"""
    import random
    status = random.choice(["active (running)", "inactive (dead)", "failed"])
    return f"{hostname} 上的 {service_name} 状态: {status}"

@tool
def search_error_log(keyword: str, log_path: str = "/var/log/syslog") -> str:
    """在日志文件中搜索包含指定关键词的错误记录"""
    return f"在 {log_path} 中找到3条包含 '{keyword}' 的记录:\n[ERROR] 2024-01-15 Connection refused\n[ERROR] 2024-01-15 Timeout exceeded\n[ERROR] 2024-01-15 Permission denied"

tools = [check_disk_usage, check_service_status, search_error_log]

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个运维助手，可以帮助用户排查服务器问题。
你有以下工具可以使用，请根据用户的问题选择合适的工具来排查。
如果需要多步排查，可以依次调用多个工具。"""),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_openai_functions_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 开启这个能看到AI的推理过程，调试必备
    max_iterations=5
)

response = agent_executor.invoke({
    "input": "帮我看看web-01服务器是不是磁盘满了，顺便查下nginx服务状态"
})
print(response["output"])

开启verbose=True之后，你能在控制台看到AI的完整推理过程——它会先分析用户的问题，决定调用哪个工具，拿到结果后再决定下一步该做什么。这个过程看着还挺有意思的，像是AI在"思考"。

但Agent有个问题需要注意：它的行为不是100%可预测的。同样的问题，不同时候可能会走不同的路径。所以在生产环境里用Agent要谨慎，最好加上一些防护措施——比如限制最大迭代次数、对工具的输入做校验、对危险操作加二次确认等等。

对话记忆

如果你做的是对话式应用，记忆管理就绕不开了。LangChain提供了几种Memory：

from langchain.memory import ConversationBufferWindowMemory

# 只记住最近10轮对话
memory = ConversationBufferWindowMemory(
    k=10,
    memory_key="chat_history",
    return_messages=True
)

为什么不用ConversationBufferMemory（记住所有对话）？因为token是有限的也是要钱的。一个长对话聊个几十轮，光历史消息就能吃掉大半的上下文窗口。WindowMemory只保留最近的N轮，简单粗暴但管用。

还有个ConversationSummaryMemory，它会用模型把历史对话压缩成摘要。听着很美好，但实际用下来发现摘要质量参差不齐，有时候重要细节会被压没了。个人感觉不如Window好用，除非你的对话真的特别长。

生产环境的一些经验

跑demo和上生产完全是两回事。我在落地过程中遇到的几个实际问题：

1. 模型调用要有降级策略

primary_llm = ChatOpenAI(model="gpt-4o")
fallback_llm = ChatOpenAI(model="gpt-4o-mini")

# 主模型挂了自动切备用
llm = primary_llm.with_fallbacks([fallback_llm])

API调用偶尔会超时或者报错，尤其是网络不稳定的时候。加个fallback能避免整个服务挂掉。

2. 监控token消耗

from langchain_community.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = rag_chain.invoke("什么是K8s的Pod？")
    print(f"本次调用：{cb.total_tokens} tokens，费用 ${cb.total_cost:.4f}")

不监控token用量，月底账单可能会给你惊喜。特别是RAG场景，每次请求都带着一堆context，token消耗比普通对话大很多。

3. 流式输出

用户等5秒才看到回答，体验非常差。流式输出能让用户感觉响应很快：

for chunk in rag_chain.stream("解释一下Docker的网络模式"):
    print(chunk, end="", flush=True)

4. 异常处理

大模型的输出不是100%可控的，JSON解析可能失败，工具调用可能出错。每个环节都要有兜底逻辑，不能让一个异常把整个服务搞崩。

我踩过的一些坑

最后说几个我实际踩过的坑，可能对你有帮助：

LangChain的版本更新很快，API经常变。网上很多教程的代码直接复制过来跑不了，因为用的是老版本的写法。建议认准官方文档，别太依赖博客教程（包括我这篇，哈哈）
langchain和langchain-community是分开的包，很多常用的loader、vectorstore都搬到community包里了
FAISS在Windows上装faiss-cpu一般没问题，但faiss-gpu很难装，别折腾了
如果你用国产模型（比如通义千问、智谱），LangChain也有对应的集成包，比如langchain-community里的ChatTongyi
embedding计算比较费时间，几千个文档片段可能要跑个几分钟，做好索引持久化别每次重算

总结

LangChain说白了就是一个AI应用开发的脚手架。你不用它也能做出同样的东西，但用了它能省很多重复劳动。它把大模型调用、文档处理、向量检索、Agent编排这些常见需求都封装好了，你只需要关注业务逻辑就行。

我的建议是：从RAG开始学，这是目前最实用也最好理解的场景。跑通一个简单的知识库问答之后，再去看Agent、Memory这些进阶的东西。别一上来就想着搞个多Agent协作系统，那玩意儿连LangChain的作者自己都还在迭代。

好了就写到这。如果你也在搞AI应用开发，或者对LangChain有什么疑问，欢迎交流。这个方向变化太快了，多交流才能跟上节奏。

如果觉得这篇文章对你有帮助，欢迎点赞、转发、在看三连，你的支持是我持续输出的动力。关注我，后续会继续分享更多AI应用开发和运维实践相关的干货内容。

公众号：耕云躬行录
个人博客：躬行笔记