背景与目标读者

LangChain 是一个面向大型语言模型（Large Language Models, LLM）应用开发的开源框架，由 Harrison Chase 于 2022 年发布，并在 2023 年成立公司后快速发展。它通过统一的抽象与模块化组件，帮助开发者高效构建复杂的 AI 应用，如聊天机器人、文档问答（RAG）、智能代理（Agent）与自动摘要等。

本文面向有一定 Python 基础、希望系统了解并快速上手 LangChain 的工程师与技术爱好者，覆盖核心组件、常见应用场景与可运行示例代码。

LangChain 是什么，为什么需要它？

  • 统一接口：屏蔽不同模型与服务的差异（如 OpenAI、Hugging Face、本地模型等），提供一致的调用方式。
  • 组件化设计：围绕模型、提示（Prompt）、链（Chain）、代理（Agent）、记忆（Memory）、索引（Indexes/Retriever）等模块化组合，便于扩展与维护。
  • 工程化能力：提供可观测（Callbacks）、持久化（Checkpointers/Message History）、工具接入（Tools）、生态集成（VectorStore/Embeddings/Loaders）等工程能力。

核心组件与用法速览

1) 模型（Models）

对接各类聊天/补全模型，常见通过 langchain-openai、langchain-community、langchain-anthropic 等包集成。

1
2
3
4
5
6
# Python
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
resp = model.invoke("用一句话解释什么是RAG？")
print(resp.content)

2) 提示（Prompts）

将系统指令、示例与用户输入组织为模板，支持变量插值与角色化消息。

1
2
3
4
5
6
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名资深AI助教，输出要简洁准确。"),
    ("human", "请用要点列举回答：{question}")
])

3) 链（Chains / LCEL）

使用 LangChain Expression Language（LCEL）以“管道”方式组合组件。

1
2
3
4
from langchain_core.output_parsers import StrOutputParser

chain = prompt | model | StrOutputParser()
print(chain.invoke({"question": "LangChain 的核心组件有哪些？"}))

4) 代理（Agents）

让 LLM 作为决策引擎，基于工具（Tools）动态选择行动，如搜索、计算或代码执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
from langchain.tools import tool
from langchain.agents import create_react_agent, AgentExecutor

@tool
def add(a: float, b: float) -> float:
    """Return a + b"""
    return a + b

tools = [add]
agent = create_react_agent(model, tools)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

print(executor.invoke({"input": "请计算 12.5 与 7.5 的和，然后用一句中文描述结果。"})["output"])

5) 记忆（Memory）

存储会话历史，使多轮对话更连贯。推荐以 RunnableWithMessageHistory 使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

history_store = {}

def get_history(session_id: str):
    return history_store.setdefault(session_id, InMemoryChatMessageHistory())

conversational_chain = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="question",
    history_messages_key="history",
)

print(conversational_chain.invoke(
    {"question": "记住我喜欢Python。"},
    config={"configurable": {"session_id": "u1"}}
))
print(conversational_chain.invoke(
    {"question": "我刚才说我喜欢哪种语言？"},
    config={"configurable": {"session_id": "u1"}}
))

6) 索引 / 检索（Indexes / VectorStores / Retrievers）

将文档切分、向量化并索引，供模型检索以提升事实性（RAG）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

docs = [
    "LangChain 提供统一的模型接口与组件化能力。",
    "RAG 将检索与生成结合，提升回答的准确性。"
]

splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
splits = splitter.create_documents(docs)

embeddings = OpenAIEmbeddings()
vectordb = FAISS.from_documents(splits, embeddings)
retriever = vectordb.as_retriever(k=2)

context_docs = retriever.invoke("什么是RAG？")
print([d.page_content for d in context_docs])

快速开始：环境与第一个程序

安装与准备

1
pip install -U "langchain>=0.2" langchain-openai langchain-community langchain-text-splitters faiss-cpu python-dotenv

配置密钥（以 OpenAI 为例），建议使用环境变量或 .env：

1
setx OPENAI_API_KEY "sk-xxxxxxxx"   # Windows PowerShell 可使用 $env:OPENAI_API_KEY="..."

Hello, LangChain（LCEL 版）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "<YOUR_KEY>")

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业的技术助理，回答务必准确且简洁。"),
    ("human", "{question}")
])

chain = prompt | model | StrOutputParser()
print(chain.invoke({"question": "一句话介绍LangChain。"}))

常见使用场景与示例

场景一：知识库问答（RAG）

将企业文档做切分与向量化，检索相关片段并与提示拼接后交给模型生成答案。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

docs = [
    "LangChain 是用于构建由大型语言模型驱动应用的框架。",
    "它包含模型、提示、链、代理、记忆与索引等组件。",
]

splitter = RecursiveCharacterTextSplitter(chunk_size=120, chunk_overlap=20)
splits = splitter.create_documents(docs)

emb = OpenAIEmbeddings()
vectordb = FAISS.from_documents(splits, emb)
retriever = vectordb.as_retriever(k=2)

prompt = ChatPromptTemplate.from_messages([
    ("system", "请基于给定上下文回答用户问题，若无法从上下文得到答案，请明确说明。上下文：\n{context}"),
    ("human", "问题：{question}")
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

def rag_answer(question: str) -> str:
    context_docs = retriever.invoke(question)
    context = "\n\n".join(d.page_content for d in context_docs)
    chain = prompt | model | parser
    return chain.invoke({"context": context, "question": question})

print(rag_answer("LangChain 的核心组件是什么？"))

场景二：多轮聊天机器人（会话记忆）

使用 RunnableWithMessageHistory 追踪对话上下文，使回答更连贯。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是友好的中文助理，记住用户偏好。"),
    ("placeholder", "{history}"),
    ("human", "{question}")
])

chain = prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0.3) | StrOutputParser()
store = {}

def get_history(sid):
    return store.setdefault(sid, InMemoryChatMessageHistory())

conv = RunnableWithMessageHistory(chain, get_history, input_messages_key="question", history_messages_key="history")

sid = "user-42"
print(conv.invoke({"question": "我喜欢Go语言，请记住。"}, config={"configurable": {"session_id": sid}}))
print(conv.invoke({"question": "我喜欢哪种语言？"}, config={"configurable": {"session_id": sid}}))

场景三：长文档摘要

将长文切分后分别摘要，再做聚合（Map-Reduce 思路）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

text = """
LangChain 通过统一抽象与模块化组件，帮助开发者构建LLM应用……（长文略）
"""

splits = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50).split_text(text)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

map_prompt = ChatPromptTemplate.from_template("请用中文简洁总结：{chunk}")
map_chain = map_prompt | model | parser

summaries = [map_chain.invoke({"chunk": c}) for c in splits]

reduce_prompt = ChatPromptTemplate.from_template("将以下摘要合并为一段清晰总结：\n{points}")
reduce_chain = reduce_prompt | model | parser

final_summary = reduce_chain.invoke({"points": "\n".join(summaries)})
print(final_summary)

场景四：工具调用/数据分析代理

通过 Agent 使用工具（如计算、搜索、数据库查询）来完成复杂任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from langchain.tools import tool
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI

@tool
def multiply(a: float, b: float) -> float:
    """Multiply two numbers"""
    return a * b

tools = [multiply]
agent = create_react_agent(ChatOpenAI(model="gpt-4o-mini"), tools)
executor = AgentExecutor(agent=agent, tools=tools)

print(executor.invoke({"input": "请计算 3.2 乘以 8，并给出中文说明。"})["output"])

进阶与工程化建议

  • 版本与依赖：
    • 推荐使用 langchain>=0.2 与分包（langchain-openai、langchain-community 等）；不同版本 API 可能有变更。
    • 向量库可选 Lancedb、FAISS、Chroma、Milvus、PGVector 等；生产建议持久化与备份策略。
  • 可观测与调试：
    • 使用 Callbacks 记录 Token 成本与时延；保留提示与上下文以便复现。
  • 安全与隐私：
    • 使用环境变量存储 API Key，不要硬编码；
    • 对数据做脱敏，遵守隐私合规；
    • 明确第三方服务的数据保留政策。
  • 部署建议：
    • 后端服务化（FastAPI 等），前后端分离；
    • 缓存热点向量检索与响应；
    • 结合队列与异步提高吞吐。

常见问题（FAQ）

  • 输出偶尔不稳定？
    • 降低 temperature，并在提示中明确约束输出格式；
    • 为 RAG 增强检索质量（更好的切分、合适的 k 值与重排序）。
  • 检索不相关？
    • 优化切分策略与嵌入模型；尝试添加领域示例与更明确的系统指令。
  • Agent 行为“走偏”？
    • 限制可用工具、添加防护指令与超时；在日志中审计推理轨迹。

参考资料

  • LangChain 官方文档（入门/指南/组件）：https://python.langchain.com/docs/
  • OpenAI 接入（langchain-openai）：https://python.langchain.com/docs/integrations/chat/openai
  • 向量数据库（FAISS/Chroma/Milvus）：https://python.langchain.com/docs/integrations/vectorstores/

本文由 AI 辅助生成，如有错误或建议，欢迎指出。