LangChain 是什么？
LangChain 是一个用于构建 LLM 应用的框架，提供了：
-
组件（Components）：可复用的 Prompt、Model、Embedding、Retriever 等
-
链（Chains）：将多个组件串联成一个处理流程
-
代理（Agents）：让 LLM 自主决策调用哪些工具
核心组件
1. Model（模型）
LangChain 支持多种 LLM：
from langchain.chat_models import ChatOpenAI
from langchain.llms import OpenAI
# ChatGPT
chat_model = ChatOpenAI(
model="gpt-4",
temperature=0.7,
max_tokens=1000
)
# GPT-4
llm = OpenAI(model="gpt-4", temperature=0.9)
# 通义千问
from langchain.chat_models import Tongyi
qwen = Tongyi(temperature=0.9)
2. Prompt（提示词）
from langchain.prompts import PromptTemplate, ChatPromptTemplate
# 简单模板
template = PromptTemplate(
input_variables=["product"],
template="为 {product} 写一个吸引人的广告语"
)
prompt = template.format(product="手机")
print(prompt) # "为 手机 写一个吸引人的广告语"
# 对话模板
chat_template = ChatPromptTemplate.from_messages([
("system", "你是一个专业的{field}顾问"),
("user", "{question}")
])
messages = chat_template.format_messages(field="法律", question="遗产继承怎么处理")
3. Chain（链）
把多个步骤串联起来：
from langchain.chains import LLMChain
chain = LLMChain(
llm=chat_model,
prompt=template
)
# 执行
result = chain.run(product="咖啡机")
print(result)
4. Memory（记忆）
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
chain = LLMChain(
llm=chat_model,
prompt=chat_prompt,
memory=memory
)
# 对话
chain.run(input="我叫张三")
chain.run(input="我叫什麼？") # 能记住"张三"
5. Retriever（检索器）
from langchain.retrievers import Retriever
# 从 VectorStore 创建 Retriever
retriever = vectorstore.as_retriever(
search_kwargs={"k": 3} # 返回Top-3
)
# 使用
docs = retriever.get_relevant_documents("什么是机器学习")
LCEL（LangChain Expression Language）
LCEL 是 LangChain 的链式调用语法，让链的编写更直观。
基础语法
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.schema import StrOutputParser
# 方式1：runnable 接口
chain = (
PromptTemplate.from_template("用一句话解释：{topic}")
| ChatOpenAI(model="gpt-4")
| StrOutputParser()
)
result = chain.invoke({"topic": "量子计算"})
print(result)
什么是 Runnable？
Runnable 是 LangChain 的核心接口，所有组件都实现了这个接口：
class Runnable:
def invoke(self, input): …
def batch(self, inputs): … # 批量
def stream(self, input): … # 流式
def pipe(self, other): … # | 操作符
常用 Runnable 组件
from langchain.schema import StrOutputParser
from langchain.runnable import RunnableParallel, RunnablePassthrough
# StrOutputParser：把输出转为字符串
# RunnablePassthrough：透传输入
# RunnableParallel：并行执行多个任务
并行与串行
from langchain.runnable import RunnableParallel
# 串行
chain1 = PromptTemplate.from_template("{topic}是什么？") | model | output
# 并行
parallel_chain = RunnableParallel(
topic_explain=chain1,
topic_examples=(
PromptTemplate.from_template("列举3个{topic}的例子") | model | output
)
)
result = parallel_chain.invoke({"topic": "人工智能"})
# result = {"topic_explain": "…", "topic_examples": "…"}
RAG 在 LangChain 中的实现
完整 RAG 链
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.schema import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
# 1. 创建 Retriever
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# 2. 组装 RAG 链
rag_chain = (
{
"two": retriever | RunnablePassthrough(),
"one": RunnablePassthrough()
}
| PromptTemplate.from_template(
"基于以下上下文回答问题。\\n"
"如果上下文不足，回答'我无法回答'。\\n\\n"
"上下文：{two}\\n\\n"
"问题：{one}"
)
| ChatOpenAI(model="gpt-4", temperature=0)
| StrOutputParser()
)
# 3. 执行
result = rag_chain.invoke("什么是机器学习？")
带Citations的RAG（可溯源）
from langchain.prompts import PromptTemplate
class RagWithCitations:
def __init__(self, vectorstore, model):
self.retriever = vectorstore.as_retriever(k=3)
self.model = model
def invoke(self, question):
# 检索
docs = self.retriever.get_relevant_documents(question)
# 组装上下文（带来源）
context_parts = []
for i, doc in enumerate(docs):
context_parts.append(f"[{i+1}] {doc.page_content}\\n来源：{doc.metadata}")
context = "\\n".join(context_parts)
# 生成回答
prompt = f"""基于以下上下文和编号回答问题。
每个编号对应一个来源，回答时用 [编号] 标注来源。
上下文：
{context}
问题：{question}
请基于上下文回答，回答时用 [编号] 标注来源。"""
response = self.model.call(prompt)
return {
"answer": response,
"sources": [doc.metadata for doc in docs]
}
自定义组件
自定义 Retriever
from langchain.schema import BaseRetriever
from langchain.schema import Document
class MyRetriever(BaseRetriever):
def __init__(self, vectorstore, top_k=5):
self.vectorstore = vectorstore
self.top_k = top_k
def get_relevant_documents(self, query):
# 自定义检索逻辑
docs = self.vectorstore.similarity_search(query, k=self.top_k)
# 自定义过滤逻辑
filtered = [doc for doc in docs if self._is_relevant(doc)]
return filtered
def _is_relevant(self, doc):
# 自定义相关性判断
return doc.metadata.get("category") == "tech"
async def aget_relevant_documents(self, query):
# 异步版本
return self.get_relevant_documents(query)
自定义 Document Transformer
from langchain.schema import Document
from langchain.text_splitter import TextSplitter
class CustomTextSplitter(TextSplitter):
def split_text(self, text):
# 自定义切分逻辑
paragraphs = text.split("\\n\\n")
chunks = []
for para in paragraphs:
if len(para) > 500:
# 大段落再细分
sentences = para.split("。")
current = ""
for sent in sentences:
if len(current) + len(sent) > 300:
if current:
chunks.append(current)
current = sent
else:
current += sent + "。"
if current:
chunks.append(current)
else:
chunks.append(para)
return [c.strip() for c in chunks if c.strip()]
自定义 Embedding
from langchain.embeddings import BaseEmbeddings
import numpy as np
class MyEmbedding(BaseEmbeddings):
def __init__(self, model_path):
from my_model import load_model
self.model = load_model(model_path)
def embed_query(self, text):
vec = self.model.encode(text)
return vec.tolist()
def embed_documents(self, texts):
vecs = self.model.encode(texts)
return vecs.tolist()
@property
def embedding_dimension(self):
return 768
Chain 的高级用法
Router Chain（路由链）
from langchain.chains import RouterChain
from langchain.prompts import PromptTemplate
# 定义多个子链
math_chain = LLMChain(llm=model, prompt=math_prompt)
history_chain = LLMChain(llm=model, prompt=history_prompt)
# 路由选择器
router_chain = RouterChain(
default_chain=general_chain,
router_chain=LLMChain(
llm=model,
prompt=RouterPromptTemplate.from_template(router_template)
),
destination_chains={
"math": math_chain,
"history": history_chain
}
)
# 自动路由
result = router_chain.run(user_input)
Sequential Chain（顺序链）
from langchain.chains import SequentialChain
# 链1：翻译
chain1 = LLMChain(
llm=model,
prompt=PromptTemplate.from_template("把以下中文翻译成英文：{text}")
)
# 链2：改写
chain2 = LLMChain(
llm=model,
prompt=PromptTemplate.from_template("把以下英文改写成更正式的版本：{text}")
)
# 串联
full_chain = SequentialChain(
chains=[chain1, chain2],
input_variables=["text"],
output_variables=["formal_text"]
)
result = full_chain.run(text="我想请假")
# chain1 输出英文 → chain2 输入 → 最终输出正式英文
输出解析器
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel
class Answer(BaseModel):
result: str
confidence: float
sources: List[str]
parser = PydanticOutputParser(pydantic_object=Answer)
chain = (
PromptTemplate.from_template("{question}")
| model
| parser
)
result = chain.invoke({"question": "什么是AI？"})
# result 是 Answer 对象，不是字符串
回调与监控
from langchain.callbacks import CallbackManager
class MyCallbackHandler(BaseCallbackHandler):
def on_llm_start(self, serialized, prompts, **kwargs):
print(f"LLM 开始处理 {len(prompts)} 个 prompt")
def on_llm_end(self, response, **kwargs):
print(f"LLM 完成，输出：{response}")
def on_chain_start(self, serialized, inputs, **kwargs):
print(f"链开始：{serialized}")
chain = LLMChain(
llm=model,
prompt=prompt,
callback_manager=CallbackManager([MyCallbackHandler()])
)
面试常问
Q: LangChain 的核心概念是什么？
A:
-
Model：LLM 调用接口
-
Prompt：提示词模板管理
-
Chain：将多个组件串联成工作流
-
Agent：让 LLM 自主决策的工具调用循环
-
Memory：对话历史管理
Q: LCEL 是什么？有什么优势？
A:
-
LCEL（LangChain Expression Language）是 LangChain 的链式调用语法
-
使用 | 操作符串联组件，如 prompt | model | output_parser
-
优势：语法简洁、支持流式、内置异步支持、方便调试
Q: LangChain 的缺点是什么？
A:
-
抽象层多，调试困难（问题定位不清晰）
-
版本迭代快，API 不稳定
-
对于简单场景，过于复杂
-
性能开销（多了很多中间层）
-
依赖过多，部署复杂
Q: 如何自定义 Retriever？
A:
-
继承 BaseRetriever
-
实现 get_relevant_documents(query) 方法
-
可以组合多个检索器（Ensemble Retriever）
-
可以加后处理过滤逻辑
