LangChain超详细2024年从零实现最小版实战教程：完整步骤与精选推荐排行榜
从零实现一个最小版 LangChain
在开始写代码之前，先弄清楚一件事：LangChain 到底是什么？一句话概括，它就是“把 LLM 应用拆成可组合组件”的框架。
它本身不生产“智能”，只提供抽象和胶水。打个比方，你就明白了：
| 传统后端 | LangChain |
| :--- | :--- |
| Spring 不发明业务逻辑，只提供 IoC/AOP 容器 | LangChain 不发明模型，只提供 LLM 调用的容器 |
| Servlet → Filter → Controller 责任链 | Prompt → LLM → Parser → Tool 责任链 |
| Bean 依赖注入 | Chain 组合（`prompt \| llm \| parser`） |
这里的关键洞察是：一个LLM应用本质上就是一条流水线：**输入变形 × 模型调用 × 输出解析 × 工具调用 × 状态管理**。而LangChain干的事情就是：把这条流水线上的每个环节，都抽成一块可以随时替换的积木。
### 核心抽象（六块积木）
来，我们先看下它的几个核心抽象，也就是那六块积木：
```
┌──────────────┐┌──────────────┐┌──────────────┐
│ PromptTemplate│──▶│ LLM │──▶│ OutputParser │
└──────────────┘└──────────────┘└──────────────┘
│ ▲
│ │
▼ │
┌──────────────┐ ┌──────────────┐
│ Memory │ │ Tool │
└──────────────┘ └──────────────┘
│
▼
┌──────────────┐
│ Agent │
└──────────────┘
```
1. **LLM**：对模型API的抽象。无论你是用OpenAI、Claude还是本地模型，接口都一样。
2. **PromptTemplate**：带变量的字符串模板，让你的提示词可以动态生成。
3. **OutputParser**：把LLM输出的字符串，解析成结构化的数据。
4. **Chain**：把上面三个串成一条流水线的核心机制。
5. **Memory**：多轮对话的状态存储，让模型不“失忆”。
6. **Tool + Agent**：让LLM能够调用外部函数，实现“思考—行动”的循环。
下面，我们就来逐个从零实现它们。
### 最小实现（可直接运行）
#### 准备：一个假的 LLM
为了代码能独立跑起来，我们先造一个“假 LLM”。真实场景里，你换成调用OpenAI或Claude的SDK就行。
```python
# file: minilangchain.py
from abc import ABC, abstractmethod
from typing import Any, Callable
import re
import json
class BaseLLM(ABC):
"""LLM 抽象基类：所有模型只暴露一个 invoke 方法"""
@abstractmethod
def invoke(self, prompt: str) -> str: ...
class FakeLLM(BaseLLM):
"""用于演示的假 LLM，按规则返回内容"""
def __init__(self, responses: dict[str, str] | None = None):
self.responses = responses or {}
def invoke(self, prompt: str) -> str:
for key, value in self.responses.items():
if key in prompt:
return value
return f"[FakeLLM echo]: {prompt[:100]}"
class OpenAILLM(BaseLLM):
"""生产环境的真实示例"""
def __init__(self, model: str = "gpt-4o-mini", api_key: str = ""):
from openai import OpenAI
self.client = OpenAI(api_key=api_key)
self.model = model
def invoke(self, prompt: str) -> str:
resp = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
)
return resp.choices[0].message.content
```
注意，这里的设计要点是：`BaseLLM` 只暴露一个 `invoke(str) -> str` 的方法。这是LangChain最底层的契约——字符串进，字符串出。其他一切都围绕这个契约做文章。
#### 积木 1：PromptTemplate
```python
class PromptTemplate:
"""带占位符的 prompt 模板。示例：PromptTemplate("你是{role}，请回答：{question}")"""
def __init__(self, template: str):
self.template = template
self.variables = re.findall(r"{(w+)}", template)
def format(self, **kwargs) -> str:
missing = set(self.variables) - set(kwargs.keys())
if missing:
raise ValueError(f"缺少变量: {missing}")
return self.template.format(**kwargs)
```
一眼就能看穿：本质上就是 `str.format()`，再加了个变量校验。LangChain 的 `PromptTemplate` 无论多复杂，核心逻辑就是这几行。
#### 积木 2：OutputParser
LLM 吐出来的是字符串，但后续代码需要的是结构化的数据（比如字典、对象）。Parser 就是干这个活儿的。
```python
class OutputParser(ABC):
@abstractmethod
def parse(self, text: str) -> Any: ...
class StrOutputParser(OutputParser):
def parse(self, text: str) -> str:
return text.strip()
class JSONOutputParser(OutputParser):
"""从 LLM 输出里提取 JSON"""
def parse(self, text: str) -> dict:
match = re.search(r"{.*}", text, re.DOTALL)
if not match:
raise ValueError(f"未找到 JSON: {text}")
return json.loads(match.group())
```
设计上的要点是：Parser 让“不可靠的字符串”变成“类型安全的数据”。这是 LangChain 能工程化的关键——在边界处做格式收敛。
#### 积木 3：Chain（核心！）
Chain 是 LangChain 的灵魂——它把 Prompt、LLM、Parser 串成一条流水线。
```python
class Runnable(ABC):
"""可被 invoke 的东西，支持用 | 组合"""
@abstractmethod
def invoke(self, input: Any) -> Any: ...
def __or__(self, other: "Runnable") -> "Chain":
"""支持 prompt | llm | parser 这种语法"""
return Chain([self, other])
class Chain(Runnable):
def __init__(self, steps: list[Runnable]):
self.steps = []
for step in steps:
# 扁平化嵌套的 Chain
if isinstance(step, Chain):
self.steps.extend(step.steps)
else:
self.steps.append(step)
def invoke(self, input: Any) -> Any:
out = input
for step in self.steps:
out = step.invoke(out)
return out
# 让之前定义的类都变成 Runnable
class PromptRunnable(Runnable):
def __init__(self, template: PromptTemplate):
self.template = template
def invoke(self, input: dict) -> str:
return self.template.format(**input)
class LLMRunnable(Runnable):
def __init__(self, llm: BaseLLM):
self.llm = llm
def invoke(self, input: str) -> str:
return self.llm.invoke(input)
class ParserRunnable(Runnable):
def __init__(self, parser: OutputParser):
self.parser = parser
def invoke(self, input: str) -> Any:
return self.parser.parse(input)
```
来看个使用示例：
```python
llm = FakeLLM({"首都": "北京"})
prompt = PromptTemplate("{country}的首都是哪里？")
chain = PromptRunnable(prompt) | LLMRunnable(llm) | ParserRunnable(StrOutputParser())
result = chain.invoke({"country": "中国"})
print(result) # -> "北京"（FakeLLM 命中规则）
```
类比后端开发，这其实就是 Unix 管道、Scala 的 `andThen` 或者 Ja va Stream 的 `map().map().map()`——函数组合。LangChain 的 LCEL 本质上就是把这种管道思想重新发明了一遍。
#### 积木 4：Memory
聊天机器人需要记住上下文，Memory 就是干这个的。
```python
class Memory:
"""最简单的对话历史管理"""
def __init__(self):
self.history: list[tuple[str, str]] = [] # [(user, ai), ...]
def add(self, user: str, ai: str):
self.history.append((user, ai))
def format(self) -> str:
"""把历史格式化进 prompt"""
if not self.history:
return ""
lines = []
for user, ai in self.history:
lines.append(f"用户: {user}")
lines.append(f"助手: {ai}")
return "n".join(lines)
class ChatChain:
"""带记忆的对话链"""
def __init__(self, llm: BaseLLM, memory: Memory | None = None):
self.llm = llm
self.memory = memory or Memory()
self.template = PromptTemplate("以下是历史对话：n{history}nn用户: {question}n助手:")
def chat(self, question: str) -> str:
prompt = self.template.format(
history=self.memory.format(),
question=question,
)
answer = self.llm.invoke(prompt)
self.memory.add(question, answer)
return answer
```
Memory 不是魔法，原理就是把历史字符串拼进 prompt 里。LangChain 里各种 Memory 的变体（滑动窗口、摘要、向量召回）本质上都是在“怎么压缩历史”这件事上做文章。
#### 积木 5：Tool + Agent（让 LLM 会用工具）
这部分是 LangChain 最“像AI”的地方：LLM 不再只是聊天，它能选择和调用函数。
##### Tool 抽象
```python
class Tool:
"""可供 LLM 调用的工具"""
def __init__(self, name: str, description: str, func: Callable[[str], str]):
self.name = name
self.description = description
self.func = func
def run(self, arg: str) -> str:
return self.func(arg)
# 示例工具
def calculator(expr: str) -> str:
try:
return str(eval(expr, {"__builtins__": {}}, {}))
except Exception as e:
return f"计算错误: {e}"
def search(query: str) -> str:
fake_db = {"天气": "北京今天晴，25℃", "股价": "苹果今日收盘 $180"}
for k, v in fake_db.items():
if k in query:
return v
return "未找到相关信息"
CALCULATOR_TOOL = Tool("calculator", "计算数学表达式，输入如 '1+2*3'", calculator)
SEARCH_TOOL = Tool("search", "搜索实时信息，输入搜索词", search)
```
##### Agent：ReAct 循环
Agent 的核心是 ReAct 模式（Reasoning + Acting）：让 LLM 在一个“思考 → 行动 → 观察”的循环里推进任务。
```python
AGENT_PROMPT = """你是一个能使用工具的 AI 助手。可用工具:
{tools}
严格按以下格式输出（每步只输出一段）:
Thought: <你的思考>
Action: <工具名>
Action Input: <工具输入>
Observation: <工具返回，由系统填入>
... (可重复多轮)
Thought: 我已得到答案
Final Answer: <最终回答>
问题: {question}
{scratchpad}"""
class Agent:
def __init__(self, llm: BaseLLM, tools: list[Tool], max_steps: int = 5):
self.llm = llm
self.tools = {t.name: t for t in tools}
self.max_steps = max_steps
def _tools_desc(self) -> str:
return "n".join(f"- {t.name}: {t.description}" for t in self.tools.values())
def run(self, question: str) -> str:
scratchpad = ""
for step in range(self.max_steps):
prompt = AGENT_PROMPT.format(
tools=self._tools_desc(),
question=question,
scratchpad=scratchpad,
)
output = self.llm.invoke(prompt)
# 终止条件：LLM 给出 Final Answer
if "Final Answer:" in output:
return output.split("Final Answer:")[-1].strip()
# 解析 Action / Action Input
action_match = re.search(r"Action:s*(w+)", output)
input_match = re.search(r"Action Input:s*(.+)", output)
if not (action_match and input_match):
return f"[Agent 解析失败]:n{output}"
tool_name = action_match.group(1).strip()
tool_input = input_match.group(1).strip()
if tool_name not in self.tools:
observation = f"错误: 工具 {tool_name} 不存在"
else:
observation = self.tools[tool_name].run(tool_input)
# 把这一轮追加进 scratchpad，给下一轮 LLM 看
scratchpad += f"n{output}nObservation: {observation}n"
return "[Agent 超出最大步数]"
```
看看使用示例：
```python
# 真实场景用 OpenAILLM，这里用 FakeLLM 模拟
fake = FakeLLM({
"scratchpad": """Thought: 我需要计算 2+3
Action: calculator
Action Input: 2+3""",
"Observation: 5": """Thought: 我已得到答案
Final Answer: 2+3=5""",
})
agent = Agent(fake, [CALCULATOR_TOOL, SEARCH_TOOL])
print(agent.run("请帮我算 2+3"))
# -> "2+3=5"
```
核心洞察是：Agent = 循环调用 LLM + 解析 LLM 指令 + 执行函数。所谓的“智能体”，本质上就是一个 `while` 循环。
#### 积木 6：Retriever + VectorStore（RAG 的基石）
RAG（检索增强生成）要解决的核心痛点是：LLM 不知道你公司的私有数据。做法是先检索相关文档片段，再把片段塞进 prompt 让 LLM 参考作答。
整个 RAG 流水线是这样的：
```
原始文档 ──► TextSplitter ──► Embeddings ──► VectorStore
│
用户问题 ──► Embeddings ──► 相似度检索 ◄───────────────┘
▼
(相关片段 + 问题) ──► LLM ──► 回答
```
##### Document & TextSplitter
```python
class Document:
"""带元数据的文本片段"""
def __init__(self, content: str, metadata: dict | None = None):
self.content = content
self.metadata = metadata or {}
class TextSplitter:
"""把长文档切成定长片段，片段之间可重叠（防止语义在边界被切断）"""
def __init__(self, chunk_size: int = 200, overlap: int = 20):
self.chunk_size = chunk_size
self.overlap = overlap
def split(self, text: str, metadata: dict | None = None) -> list[Document]:
chunks = []
start = 0
while start < len(text):
end = min(start + self.chunk_size, len(text))
chunks.append(Document(text[start:end], metadata))
if end == len(text):
break
start = end - self.overlap
return chunks
```
##### VectorStore（用 Jaccard 相似度模拟向量检索）
真实的 VectorStore 会用 Embedding 模型把文本转成高维向量，然后用余弦相似度检索。这里我们用 Jaccard 相似度（词集合交并比）来代替，这样能把抽象讲清楚，又不需要引入 numpy 依赖。
```python
class VectorStore:
"""最简实现。真实版本:
- add() 时调用 Embeddings 模型把文本转向量，存到 FAISS/Chroma/Milvus
- search() 时把 query 也转向量，用余弦相似度 top-k
这里用 Jaccard 替代向量相似度,便于零依赖运行。"""
def __init__(self):
self.docs: list[Document] = []
def add(self, docs: list[Document]):
self.docs.extend(docs)
@staticmethod
def _tokenize(text: str) -> set[str]:
# 真实场景这里是 embedding 向量化
return set(re.findall(r"w+", text.lower()))
def search(self, query: str, k: int = 3) -> list[Document]:
q_tokens = self._tokenize(query)
if not q_tokens:
return []
scored = []
for doc in self.docs:
d_tokens = self._tokenize(doc.content)
if not d_tokens:
continue
score = len(q_tokens & d_tokens) / len(q_tokens | d_tokens)
scored.append((score, doc))
scored.sort(key=lambda x: x[0], reverse=True)
return [doc for _, doc in scored[:k] if _ > 0]
```
##### Retriever（Runnable 化，可插入 Chain）
```python
class Retriever(Runnable):
"""把 VectorStore 包装成 Runnable，可用 | 串进 Chain"""
def __init__(self, vectorstore: VectorStore, k: int = 3):
self.vectorstore = vectorstore
self.k = k
def invoke(self, query: str) -> str:
docs = self.vectorstore.search(query, self.k)
if not docs:
return "（未检索到相关内容）"
return "n---n".join(d.content for d in docs)
```
设计上的要点是：Retriever 的输入是字符串（问题），输出也是字符串（拼好的上下文）——这完全符合 Runnable 的契约，所以它可以无缝地插入到 `|` 管道中。这是 LangChain 抽象最优雅的地方：RAG 不是什么特殊的东西，它只是管道里多加了一截水管。
### 把所有积木拼起来：一个完整的 RAG + Agent 应用
```python
class MiniLangChain:
"""所有抽象的集大成者"""
def __init__(self, llm: BaseLLM):
self.llm = llm
self.memory = Memory()
self.tools: list[Tool] = []
self.vectorstore = VectorStore()
self.splitter = TextSplitter(chunk_size=100, overlap=10)
def add_tool(self, tool: Tool):
self.tools.append(tool)
def ingest(self, text: str, metadata: dict | None = None):
"""把文档切片入库,为 RAG 做准备"""
docs = self.splitter.split(text, metadata)
self.vectorstore.add(docs)
def chain(self, template: str, parser: OutputParser | None = None):
parser = parser or StrOutputParser()
return (
PromptRunnable(PromptTemplate(template))
| LLMRunnable(self.llm)
| ParserRunnable(parser)
)
def rag_chain(self, k: int = 3):
"""构建 RAG 链: 问题 -> 检索 -> 拼 prompt -> LLM -> 回答"""
retriever = Retriever(self.vectorstore, k=k)
rag_prompt = PromptTemplate(
"基于以下上下文回答问题。若上下文没有答案,请回答"不知道"。nn"
"上下文:n{context}nn问题: {question}n回答:"
)
def invoke(question: str) -> str:
context = retriever.invoke(question)
prompt = rag_prompt.format(context=context, question=question)
return self.llm.invoke(prompt)
return invoke
def agent(self) -> Agent:
return Agent(self.llm, self.tools)
def chat(self, question: str) -> str:
return ChatChain(self.llm, self.memory).chat(question)
# ============ 完整使用示例 ============
fake = FakeLLM({
"LangChain": "LangChain 是一个让 LLM 应用组件化的 Python 框架",
"首都": "北京",
})
app = MiniLangChain(fake)
# 1. 简单链
summarizer = app.chain("请用一句话总结: {text}")
print(summarizer.invoke({"text": "关于 LangChain 的长篇介绍..."}))
# 2. RAG: 先喂知识，再问
app.ingest("""LangChain 是 2022 年由 Harrison Chase 创建的开源项目。
它的核心抽象包括 LLM、PromptTemplate、Chain、Memory、Tool、Agent。
LangChain 的配套产品 LangSmith 提供 tracing 和调试能力。
后来推出的 LangGraph 用状态机替代了线性 Chain,更适合复杂 Agent。""")
rag = app.rag_chain(k=2)
print(rag("LangChain 是什么？"))
# LLM 会基于检索到的上下文回答,而不是瞎编
# 3. 带记忆的对话
app.chat("你好")
app.chat("我刚才说了什么？")
# 会记住上下文
# 4. Agent 使用工具
app.add_tool(CALCULATOR_TOOL)
app.add_tool(SEARCH_TOOL)
app.agent().run("今天北京天气怎么样？然后帮我算 23*17")
```
到这里，你其实已经手工搭出了一个能做 RAG、对话、工具调用的 LLM 应用框架。所有概念都压缩在大约 300 行代码里，而且每一行都在你的控制之下。
### LangChain 真实代码 vs 我们的最小版
| 抽象 | 最小版 | LangChain 真实实现 |
| :--- | :--- | :--- |
| `BaseLLM` | 1 个方法 | `BaseLanguageModel` + `BaseChatModel`，支持流式、批量、异步、回调 |
| `PromptTemplate` | 正则替换 | 支持 few-shot、partial、MessagesPlaceholder、多模态 |
| `Chain` | `\|` 串联 | LCEL，支持并行（`RunnableParallel`）、分支、重试、fallback |
| `Memory` | list 拼接 | `BufferMemory`、`SummaryMemory`、`VectorStoreMemory`（向量召回） |
| `Tool` | 函数包装 | 支持 Pydantic 参数 Schema、异步、权限控制 |
| `Agent` | 字符串解析 ReAct | 支持 function calling、OpenAI tools、LangGraph 状态机 |
| `Retriever` / `VectorStore` | Jaccard 相似度词袋 | 真实 Embeddings (OpenAI/BGE) + FAISS/Chroma/Milvus + 多种检索策略 (MMR、Hybrid、Re-rank) |
结论很明确：LangChain 本质上没什么神秘的——它是在这 6 块积木上加了 100 种变体，外加对几十家模型、向量库、工具的适配器。
### 用真实 LangChain 重写上面的示例
下面我们把第四节里那个“RAG + 对话 + Agent”应用，用真实的 LangChain 重写一遍。对照着看，你会立刻明白“我们自己搭的 MiniLangChain”和“真货”之间的映射关系。
#### 环境准备
```bash
pip install langchain langchain-openai langchain-community langchain-chroma langchain-text-splitters langgraph
export OPENAI_API_KEY=sk-...
```
#### 示例 1：简单链 (对应我们的 `app.chain(...)`)
```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template("请用一句话总结: {text}")
# 这就是 LCEL: 和我们 MiniLangChain 里的 | 组合一模一样
summarizer = prompt | llm | StrOutputParser()
print(summarizer.invoke({"text": "LangChain 是 LLM 应用的组件化框架..."}))
```
映射关系很清晰：
| 我们的最小版 | 真实 LangChain |
| :--- | :--- |
| `PromptRunnable(PromptTemplate(...))` | `ChatPromptTemplate.from_template(...)` |
| `LLMRunnable(OpenAILLM(...))` | `ChatOpenAI(...)` |
| `ParserRunnable(StrOutputParser())` | `StrOutputParser()` |
| 我们的 `\|` 操作符 | 一模一样的 `\|` (LCEL) |
#### 示例 2：RAG (对应我们的 `app.rag_chain(...)`)
真实 LangChain 的 RAG 要比我们的 MiniLangChain 多两步：真实的 Embeddings 模型和真实的向量库。
```python
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 1. 切片(对应 TextSplitter)
splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=10)
docs = splitter.create_documents(["""LangChain 是 2022 年由 Harrison Chase 创建的开源项目。
它的核心抽象包括 LLM、PromptTemplate、Chain、Memory、Tool、Agent。
LangChain 的配套产品 LangSmith 提供 tracing 和调试能力。
后来推出的 LangGraph 用状态机替代了线性 Chain,更适合复杂 Agent。"""])
# 2. 向量化 + 入库(对应 VectorStore.add)
# OpenAIEmbeddings 会调 API 把每个 chunk 转成 1536 维向量
# Chroma 是本地向量库,生产环境可换 FAISS/Milvus/Pinecone
vectorstore = Chroma.from_documents(docs, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
# 3. 组装 RAG 链(对应 rag_chain)
rag_prompt = ChatPromptTemplate.from_template(
"基于以下上下文回答问题。若上下文没有答案,请回答"不知道"。nn"
"上下文:n{context}nn问题: {question}"
)
def format_docs(docs):
return "n---n".join(d.page_content for d in docs)
# 这个 dict 写法是 LCEL 的并行分发: 同一个输入分别流向 context 和 question 两路
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| rag_prompt
| ChatOpenAI(model="gpt-4o-mini")
| StrOutputParser()
)
print(rag_chain.invoke("LangChain 是什么?"))
```
这里有几个关键差异值得注意：
| 我们的最小版 | 真实 LangChain |
| :--- | :--- |
| Jaccard 相似度(词集合交并比) | OpenAI Embeddings + 余弦相似度(1536 维向量) |
| Python list 存文档 | Chroma(本地)/FAISS/Milvus/Pinecone |
| `rag_chain()` 返回函数 | LCEL 管道(原生支持流式、异步、batch) |
| 手动拼 context 字符串 | LCEL `dict` 并行分发 + `RunnablePassthrough` |
#### 示例 3：带记忆的对话 (对应我们的 `app.chat(...)`)
```python
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个乐于助人的助手"),
MessagesPlaceholder(variable_name="history"), # 历史消息插入点
("human", "{question}"),
])
chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
# 按 session_id 隔离不同用户的对话历史
store: dict[str, InMemoryChatMessageHistory] = {}
def get_history(session_id: str) -> InMemoryChatMessageHistory:
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
chat_with_memory = RunnableWithMessageHistory(
chain,
get_history,
input_messages_key="question",
history_messages_key="history",
)
# 注意 config 里的 session_id - 这是 LangChain 多租户的核心
cfg = {"configurable": {"session_id": "user_42"}}
print(chat_with_memory.invoke({"question": "你好,我叫杰克"}, config=cfg))
print(chat_with_memory.invoke({"question": "我刚才说了什么？"}, config=cfg))
# -> 会正确回答"你说你叫杰克"
```
关键差异一览：
| 我们的最小版 | 真实 LangChain |
| :--- | :--- |
| 单例 `Memory` 存在内存 list | `RunnableWithMessageHistory` + 可插拔 `ChatMessageHistory`(Redis/Postgres/DynamoDB) |
| 手动拼 prompt 字符串 | `MessagesPlaceholder` 自动注入 |
| 无多用户隔离 | `session_id` 多租户开箱即用 |
#### 示例 4：Agent (对应我们的 `app.agent()...`)
现代 LangChain 推荐用 LangGraph 的 `create_react_agent`，而不是旧的 `AgentExecutor`。LangGraph 把 Agent 建模成状态图，比我们的 `while` 循环强大得多（支持分支、并行、人在回路、checkpoint 回放）。
```python
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
@tool
def calculator(expr: str) -> str:
"""计算一个数学表达式,输入如 '2+3*5'"""
try:
return str(eval(expr, {"__builtins__": {}}, {}))
except Exception as e:
return f"计算错误: {e}"
@tool
def search(query: str) -> str:
"""搜索实时信息,输入搜索词"""
fake_db = {"天气": "北京今天晴,25℃", "股价": "苹果今日收盘 $180"}
for k, v in fake_db.items():
if k in query:
return v
return "未找到"
# 一行代码构建 ReAct Agent
agent = create_react_agent(
ChatOpenAI(model="gpt-4o-mini"),
tools=[calculator, search],
)
# 注意输入输出格式是消息列表
result = agent.invoke({
"messages": [("user", "今天北京天气怎么样？然后帮我算 23*17")]
})
print(result["messages"][-1].content)
```
关键差异：
| 我们的最小版 | 真实 LangChain |
| :--- | :--- |
| 字符串正则解析 `Action:` / `Action Input:` | 原生 function calling(OpenAI/Anthropic 模型直接吐结构化 tool call) |
| 手写 ReAct prompt | LangGraph 内置 prompt + 状态机 |
| `while` 循环手工推进 | 状态图调度,可中断、可回放、可并行工具调用 |
| `Tool(name, description, func)` | `@tool` 装饰器自动从函数签名和 docstring 提取 schema |
| 无 Schema 校验 | 自动生成 Pydantic schema,输入类型校验 |
#### 示例 5：把它们串成一个生产可用的应用
上面 4 个示例在生产里常常需要组合起来：带记忆 + 有 RAG 检索 + 能调工具的 Agent。这是真实业务里最常见的形态。
```python
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySa ver
from langchain_core.tools import tool
# 把 retriever 包装成 tool,让 Agent 自己决定什么时候查知识库
@tool
def lookup_knowledge(query: str) -> str:
"""查询公司内部知识库"""
docs = retriever.invoke(query)
return "n---n".join(d.page_content for d in docs)
# checkpointer = Agent 的"记忆"(状态持久化)
agent = create_react_agent(
ChatOpenAI(model="gpt-4o-mini"),
tools=[calculator, search, lookup_knowledge],
checkpointer=MemorySa ver(), # 生产环境换 PostgresSa ver/RedisSa ver
)
cfg = {"configurable": {"thread_id": "conversation_1"}}
# 多轮对话,Agent 自己决定什么时候 RAG、什么时候算数、什么时候搜
agent.invoke({"messages": [("user", "LangChain 核心组件有哪些？")]}, config=cfg)
agent.invoke({"messages": [("user", "那你刚才提到的第三个是什么？")]}, config=cfg)
```
这就是“LangChain 全家桶”的真实形态：LCEL 做管道、LangGraph 做 Agent、LangSmith 做 tracing（下一节会提）。
#### 整体对照表
| 能力 | MiniLangChain(~300 行) | 真实 LangChain |
| :--- | :--- | :--- |
| 简单链 | `prompt \| llm \| parser` | `prompt \| llm \| parser`(同款 LCEL) |
| RAG | Jaccard + list | OpenAIEmbeddings + Chroma/FAISS + LCEL 分发 |
| Memory | 单例 list 拼字符串 | `RunnableWithMessageHistory` + 多后端 |
| Agent | 正则解析 ReAct 循环 | LangGraph 状态机 + native function calling |
| Tracing / 调试 | `print` | LangSmith(DAG 可视化、token 计费、回放) |
| 异步 / 流式 / batch | 无 | 全部 Runnable 原生支持 |
| 模型切换 | 换个 `BaseLLM` 子类 | 换 `langchain-*` 包的导入路径,其余不动 |
看完这节应该能明白：你自己写的 MiniLangChain 在抽象层面和真实 LangChain 是同构的——都是 Runnable 组合。真实 LangChain 多出来的价值，90% 在适配器生态（对接 50+ 模型、30+ 向量库、100+ 工具）和周边工程能力（LangSmith tracing、流式、异步、checkpoint），而不在核心抽象。
这也是为什么建议“先裸写再决定用不用”——你现在已经有能力做出判断了。
### 给后端工程师的理解捷径
如果你来自 Spring / Scala / Ja va 生态，可以这样映射：
| Spring 概念 | LangChain 对应 |
| :--- | :--- |
| `@Component` Bean | `Runnable` |
| `BeanFactory` | `LLM` provider |
| Filter Chain | LCEL Chain |
| `@Cacheable` | `set_llm_cache` |
| AOP 切面 | Callback Handler |
| `ApplicationContext` | Agent（持有一堆 tools） |
| Spring Integration 的 EIP | LangGraph 的 state machine |
核心的范式转变在于：
- **传统后端**：确定性流程 + 偶尔的不确定输入
- **LLM 应用**：不确定性组件 + 确定性的编排层
LangChain 做的就是把不确定性（LLM 输出）封在 `Runnable.invoke()` 里，让外层保持工程师熟悉的确定性组合模式。
### 什么时候不需要 LangChain？
坦诚地说，LangChain 在 2024 年之后争议很大。
**适合用它的场景：**
- 需要快速切换多个模型供应商（抽象的价值就体现出来了）
- 处理复杂的多步骤 Agent，需要现成的 Memory / Retriever 实现
- 需要 LangSmith 做 tracing 和调试
**不适合用它的场景：**
- 只用单一模型（比如只用 OpenAI），直接调 SDK 更简单
- 对延迟和代码可读性要求高——LangChain 抽象层深，栈跟踪看起来像噩梦
- 复杂的