LangChain 核心概念与组件-开发者社区

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 的核心概念是什么？

Model：LLM 调用接口
Prompt：提示词模板管理
Chain：将多个组件串联成工作流
Agent：让 LLM 自主决策的工具调用循环
Memory：对话历史管理

Q: LCEL 是什么？有什么优势？

LCEL（LangChain Expression Language）是 LangChain 的链式调用语法
使用|操作符串联组件，如prompt | model | output_parser
优势：语法简洁、支持流式、内置异步支持、方便调试

Q: LangChain 的缺点是什么？

抽象层多，调试困难（问题定位不清晰）
版本迭代快，API 不稳定
对于简单场景，过于复杂
性能开销（多了很多中间层）
依赖过多，部署复杂

Q: 如何自定义 Retriever？

继承 BaseRetriever
实现get_relevant_documents(query)方法
可以组合多个检索器（Ensemble Retriever）
可以加后处理过滤逻辑

LangChain 核心概念与组件