|
LangChain 是"大模型应用编排框架"：把 **模型 / 提示词 / 输出解析 / 检索 / 记忆 / 工具** 抽象成可组合的组件。
```bash
pip install langchain langchain-core langchain-openai langchain-community langchain-classic
```
1.x 时代包结构：**langchain-core**（抽象）/ **langchain-classic**（链与记忆）/ **langchain-community**（社区集成）/ **langchain-openai** 等独立集成包。
密钥统一走环境变量（禁止硬编码进代码）：
```python
import os
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv(filename="D:/wks/jupyter/env.txt"))
model_name = os.getenv("deepseek_model")
base_url = os.getenv("deepseek_base_url")
api_key = os.getenv("deepseek_api_key")
```
|
|
ChatOpenAI 兼容一切 OpenAI 协议端点（deepseek/通义/火山/本地 vllm 均可）：
```python
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model=model_name, openai_api_key=api_key,
openai_api_base=base_url, temperature=0.7, max_tokens=1000)
resp = llm.invoke("用一句话介绍LangChain")
print(resp.content)
print(resp.usage_metadata) # token 用量：input/output/total
```
对话三消息类型（对应 role: system/user/assistant）：
```python
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
messages = [
SystemMessage(content="你是专业AI客服小达，回答简洁"),
HumanMessage(content="你好，我是张同学"),
AIMessage(content="张同学你好！有什么可以帮您？"),
HumanMessage(content="Python全栈课程要学多久？"),
]
print(llm.invoke(messages).content)
```
|
|
流式输出（打字机效果）：
```python
from langchain_core.callbacks import StreamingStdOutCallbackHandler
sm = ChatOpenAI(model=model_name, openai_api_key=api_key, openai_api_base=base_url,
streaming=True, callbacks=[StreamingStdOutCallbackHandler()])
sm.invoke("用50字写一个科幻开头") # 逐 token 打印
```
批量与容灾降级：
```python
llm.batch(["你好", "介绍Python"]) # 批量并行调用
backup = ChatOpenAI(model=model_name, openai_api_key=api_key, openai_api_base=base_url)
llm_ha = primary_llm.with_fallbacks([backup]) # 主模型异常自动切备用
llm_robust = llm.with_retry(stop_after_attempt=3) # 瞬时错误指数退避重试
```
**初学者提示**：先用一个模型把链路调通，再考虑容灾/限流（`InMemoryRateLimiter`）与缓存（`set_llm_cache`）。
|
|
把提示词从"硬编码字符串"升级为"可复用模板"：
```python
from langchain_core.prompts import PromptTemplate
ad = PromptTemplate.from_template("""
你是一位资深广告文案策划师。请为以下产品创作文案：
产品名称：{product_name}
文案风格：{style}
""")
print(ad.input_variables) # ['product_name', 'style']
p = ad.format(product_name="扫地机器人", style="科技感")
resp = llm.invoke(p) # 模板.format() 后直接喂模型
```
也可从文件加载模板：`load_prompt("simple_prompt.json")`（支持 yaml/json）。
|
|
多消息 + 角色设定（对话系统标配）：
```python
from langchain_core.prompts import (
ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate)
tpl = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是{language}专家，专长{specialties}，回答不超过50字"),
HumanMessagePromptTemplate.from_template("{question}"),
])
msgs = tpl.format_messages(language="Python", specialties="Web开发",
question="装饰器是什么？")
print(llm.invoke(msgs).content)
```
`MessagesPlaceholder("history")` 可在模板中预留"任意长度消息列表"的位置（多轮对话的关键）：
```python
from langchain_core.prompts import MessagesPlaceholder
pt = ChatPromptTemplate.from_messages([
("system", "你是有用助手"),
MessagesPlaceholder("history"), # 等价简写：("placeholder", "{history}")
("human", "{question}"),
])
```
|
|
少样本学习：给 AI 几个例子，比写规则更直观：
```python
from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
example_prompt = PromptTemplate.from_template("问: {question}\n{answer}")
examples = [
{"question": "谁活得更久，李小龙还是爱因斯坦？",
"answer": "逐步推理略。所以最终答案是: 爱因斯坦"},
]
fs = FewShotPromptTemplate(examples=examples, example_prompt=example_prompt,
suffix="问题: {input}", input_variables=["input"])
print(fs.invoke({"input": "谁是李白的老师？"}).to_string())
```
**动态选样例**（语义相似度，需 embedding，见 L2-2）：
```python
from langchain_chroma import Chroma
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
selector = SemanticSimilarityExampleSelector.from_examples(
examples, embeddings, Chroma, k=2) # 按输入自动挑最相似的 2 个示例
fs2 = FewShotPromptTemplate(example_selector=selector, example_prompt=example_prompt,
suffix="问题: {input}", input_variables=["input"])
```
|
|
最简单的解析器：把 AIMessage 转成纯文本，LCEL 管道末端标配：
```python
from langchain_core.output_parsers import StrOutputParser
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"topic": "程序员"}) # 直接是 str，不是 AIMessage
```
|
|
让 LLM 输出**带校验的结构化数据**：pydantic 定 schema → format_instructions 注入提示词 → parse 成强类型对象：
```python
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from typing import List, Optional
class Person(BaseModel):
name: str = Field(description="姓名")
skills: List[str] = Field(description="技能列表")
work_years: Optional[int] = Field(description="工作年限")
parser = PydanticOutputParser(pydantic_object=Person)
pt = PromptTemplate(
template="从文本提取人员信息。\n文本：{text}\n{format_instructions}\n只输出JSON。",
input_variables=["text"],
partial_variables={"format_instructions": parser.get_format_instructions()},
)
chain = pt | llm | parser
p = chain.invoke({"text": "张三，8年经验，会Python和Go"})
print(p.name, p.skills, p.work_years) # 强类型对象，直接做业务处理
```
|
|
模型支持 tool calling 时的**现代首选**（原生结构化通道，合规率更高）：
```python
class Joke(BaseModel):
setup: str = Field(description="笑话铺垫")
punchline: str = Field(description="笑点")
joke = llm.with_structured_output(Joke).invoke("讲个程序员笑话")
print(joke.setup, joke.punchline) # 直接是 Joke 对象
```
⚠️ **deepseek thinking 模型实测坑**（deepseek-v4-flash）：
```python
# 必须 function_calling 模式 + 关闭思考，缺一个都报 400
llm_fc = ChatOpenAI(model=model_name, openai_api_key=api_key, openai_api_base=base_url,
extra_body={"thinking": {"type": "disabled"}})
structured = llm_fc.with_structured_output(Joke, method="function_calling")
```
解析失败可自我修复：`OutputFixingParser.from_llm(parser=base_parser, llm=llm)` 会自动让 LLM 修复不合规输出。
|
|
所有 Loader 统一返回 `List[Document]`（page_content + metadata）：
```python
from langchain_community.document_loaders import (
TextLoader, PyPDFLoader, CSVLoader, WebBaseLoader,
UnstructuredMarkdownLoader, UnstructuredWordDocumentLoader)
docs = TextLoader("a.txt", encoding="utf-8").load() # 中文务必指定编码
pages = PyPDFLoader("b.pdf").load() # 每页一个 Document
rows = CSVLoader("c.csv").load() # 每行一个 Document
web = WebBaseLoader("https://example.com").load() # 网页
```
工程模式：按扩展名路由到对应 Loader，批量扫目录收集进知识库（`Path.rglob` + 异常逐个捕获）。
|
|
分割决定检索质量。五种常用：
```python
from langchain_text_splitters import (
RecursiveCharacterTextSplitter, # 通用首选：按 ["\n\n","\n"," ",""] 递归尝试
MarkdownHeaderTextSplitter, # md 按标题层级切，标题进 metadata
TokenTextSplitter, # 按 token 数（贴模型输入限制）
PythonCodeTextSplitter, # 代码文件
CharacterTextSplitter) # 单分隔符
splitter = RecursiveCharacterTextSplitter(
chunk_size=300, chunk_overlap=50, length_function=len)
chunks = splitter.split_documents(docs)
```
实测坑：**自定义 chunk_size 要显式给 chunk_overlap**（默认 200，比 chunk_size 小会直接报错）。
参数语义：先保证单段不超 chunk_size，再把相邻段合并到上限；overlap 保持上下文连续。
|
|
中文长文建议"段落优先、超长按句切"：
```python
import re
class SmartChineseSplitter:
def __init__(self, chunk_size=200, chunk_overlap=30):
self.chunk_size = chunk_size
def split_text(self, text):
chunks, current = [], ""
for para in text.split("\n\n"):
if len(current + para) <= self.chunk_size:
current += para + "\n\n"
else:
if current.strip(): chunks.append(current.strip())
current = para + "\n\n"
if current.strip(): chunks.append(current.strip())
return chunks
def _split_by_sentence(self, text):
return [s.strip() + "。" for s in re.split(r"[。！？；\n]", text) if s.strip()]
```
质量评估：统计块数/平均长度/长度范围，避免过碎或过长（块大小要适配 embedding 模型输入上限）。
|
|
云端（OpenAI 兼容端点，示例为 DashScope 兼容模式）：
```python
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model="qwen3.7-text-embedding",
openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
openai_api_base=os.getenv("DASHSCOPE_URL_BEIJING"),
check_embedding_ctx_length=False, # 非官方端点必加，跳过本地 tiktoken 预检
)
```
本地（数据不出内网/零调用成本）：
```python
from langchain_huggingface import HuggingFaceEmbeddings
local_emb = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5", # 首次下载走镜像：HF_ENDPOINT=https://hf-mirror.com
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True})
```
大批量重复向量化用 `CacheBackedEmbeddings` 包一层缓存。
|
|
```python
from langchain_community.vectorstores import Chroma, FAISS
from langchain_core.documents import Document
docs = [Document(page_content=t, metadata={"id": i}) for i, t in enumerate(texts)]
# Chroma：易用 + 持久化
vs = Chroma.from_documents(docs, embeddings,
collection_name="knowledge_base", # ⚠️ 集合名至少 3 字符
persist_directory="./chroma_db")
# FAISS：本地高性能，文件保存/加载
vfs = FAISS.from_documents(docs, embeddings)
vfs.save_local("./faiss_index")
vfs = FAISS.load_local("./faiss_index", embeddings,
allow_dangerous_deserialization=True)
```
⚠️ 实测坑：持久化目录里的 collection 绑定 embedding 维度，换模型必须先删库重建（否则报维度不一致）；Windows 下运行中删 chroma 目录会被文件锁挡住——启动时预清理，运行后不删。
|
|
```python
retriever = vs.as_retriever(search_kwargs={"k": 3}) # 检索器抽象（LCEL 可组合）
docs = retriever.invoke("什么是RAG？")
vs.similarity_search("query", k=2) # 1. 相似度
vs.similarity_search_with_score("query", k=2) # 2. 带距离分（越小越相似）
vs.max_marginal_relevance_search("query", k=3, fetch_k=6) # 3. MMR 去重保多样性
vs.similarity_search_with_relevance_scores("query", score_threshold=0.7)
# 4. 阈值过滤（0~1 相关度，越大越相似——注意与距离分方向相反）
```
经验：期望答案可能排在第 2、3 位——严格场景把 Top-K 全送 LLM 判断，别只信第 1 名。
|
|
检索增强生成 = 检索相关知识 + 塞进提示词 + LLM 基于上下文回答：
```python
from langchain_classic.chains import RetrievalQA
from langchain_core.prompts import PromptTemplate
rag_prompt = PromptTemplate.from_template("""
基于以下上下文回答问题。上下文中没有就明说。
上下文：{context}
问题：{question}
""")
qa = RetrievalQA.from_chain_type(
llm=llm, chain_type="stuff", # 检索块直接塞进 prompt
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
chain_type_kwargs={"prompt": rag_prompt},
return_source_documents=True, # 引用溯源
)
r = qa.invoke({"query": "年假怎么规定的？"})
print(r["result"], r["source_documents"]) # 答案 + 来源
```
|
|
1.x 推荐写法（管道组合，自动获得 batch/stream/async）：
```python
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
def format_docs(docs):
return "\n\n".join(d.page_content for d in docs)
rag_chain = (
RunnableParallel({"context": retriever | format_docs,
"question": RunnablePassthrough()})
| rag_prompt | llm | StrOutputParser()
)
print(rag_chain.invoke("病假工资怎么算？"))
```
|
|
效果不佳时按层叠加优化：
```python
from langchain_classic.retrievers import BM25Retriever, EnsembleRetriever
from langchain_classic.retrievers import ContextualCompressionRetriever
from langchain_classic.retrievers.document_compressors import LLMChainExtractor
# 1. 混合检索：BM25(关键词) + 向量(语义) 加权融合
bm25 = BM25Retriever.from_documents(docs); bm25.k = 3
ens = EnsembleRetriever(retrievers=[bm25, vec_retriever], weights=[0.4, 0.6])
# 2. LLM 重排序：从召回文档中抽取真正相关部分，降噪
comp = ContextualCompressionRetriever(
base_compressor=LLMChainExtractor.from_llm(llm), base_retriever=ens)
```
进阶（按需取用）：**查询重写**（一 query 扩多个子查询多路召回去重）、**ParentDocumentRetriever**（小块检索、大块返回）、**Hyde**（先生成假设答案再检索）、**RAG-Fusion**（多查询 + RRF 融合排序）、**元数据过滤**。
|
|
| 记忆 | 类 | 特点 |
|------|----|----|
| 缓冲 | ConversationBufferMemory | 全量保留，短对话 |
| 窗口 | ConversationBufferWindowMemory(k=2) | 只留最近 k 轮，控 token |
| 摘要 | ConversationSummaryMemory(llm=llm) | LLM 压缩历史，超长对话 |
| 实体 | ConversationEntityMemory(llm=llm) | 抽取人名/组织等实体 |
| 向量 | VectorStoreRetrieverMemory | 语义检索相关历史 |
| 组合 | CombinedMemory | 多记忆并行 |
```python
from langchain_classic.memory import ConversationBufferMemory
from langchain_classic.chains import ConversationChain
conv = ConversationChain(llm=llm, memory=ConversationBufferMemory())
conv.predict(input="你好，我叫小明")
print(conv.predict(input="我叫什么名字？")) # 能记住"小明"
```
注意：Summary/Entity 记忆每轮要额外消耗 LLM 调用。
|
|
海量历史按需语义检索（只把相关历史拼进 prompt）：
```python
from langchain_classic.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
vs = Chroma(embedding_function=embeddings)
vmem = VectorStoreRetrieverMemory(
retriever=vs.as_retriever(search_kwargs=dict(k=2)),
memory_key="history", input_key="input")
vmem.save_context({"input": "讨论微服务架构"}, {"output": "微服务是拆分小服务的设计"})
relevant = vmem.load_memory_variables({"input": "微服务有什么建议？"}) # 只检索相关历史
```
接入对话链即闭环：`ConversationChain(llm=llm, memory=vmem)`——提问→检索相关历史→生成→自动回写。
|
|
内存记忆进程退出即丢失；用 PostgreSQL 持久化（多会话隔离、跨进程恢复）：
```python
import uuid
from langchain_community.chat_message_histories import PostgresChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
DATABASE_URL = "postgresql://user:password@host:5432/testdb"
sid = "user_" + str(uuid.uuid4())[:8]
hist = PostgresChatMessageHistory(connection_string=DATABASE_URL, session_id=sid)
hist.add_user_message("你好！我是张三") # 自动建 message_store 表
prompt = ChatPromptTemplate.from_messages([
("system", "你是有用助手，记住对话历史"),
MessagesPlaceholder("history"),
("human", "{question}"),
])
chain = prompt | llm
cwh = RunnableWithMessageHistory(
chain, lambda s: PostgresChatMessageHistory(connection_string=DATABASE_URL, session_id=s),
input_messages_key="question", history_messages_key="history")
cfg = {"configurable": {"session_id": sid}}
cwh.invoke({"question": "我喜欢打篮球"}, config=cfg)
r = cwh.invoke({"question": "我喜欢什么运动？"}, config=cfg) # 跨轮记忆
```
|
|
LCEL：一切组件都是 Runnable，用 `|` 像 Unix 管道组合；组合后免费获得全套执行接口：
```python
chain = prompt | llm | StrOutputParser()
chain.invoke({"topic": "程序员"}) # 单次
chain.batch([{"topic": t} for t in ["设计师", "产品经理"]]) # 批量
for chunk in chain.stream({"topic": "猫"}): # 流式
print(chunk, end="|")
# await chain.ainvoke(...) # 异步
```
|
|
RunnableParallel：多个子链**并行**执行同一输入，总耗时≈最慢子链：
```python
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
analysis = RunnableParallel(
sentiment=sentiment_prompt | llm | StrOutputParser(),
summary=summary_prompt | llm | StrOutputParser(),
keywords=keyword_prompt | llm | StrOutputParser(),
original=RunnablePassthrough(), # 透传原始输入
)
r = analysis.invoke({"text": "LangChain让AI应用开发更简单"})
# r = {"sentiment": ..., "summary": ..., "keywords": ..., "original": ...}
```
|
|
RunnableBranch：按输入内容动态路由到不同处理链：
```python
from langchain_core.runnables import RunnableBranch
def classify(x):
return "tech" if any(k in x["q"] for k in ["代码", "python", "api"]) else "gen"
router = RunnableBranch(
(lambda x: classify(x) == "tech", tech_prompt | llm | StrOutputParser()),
(lambda x: classify(x) == "creative", creative_prompt | llm | StrOutputParser()),
general_prompt | llm | StrOutputParser(), # 默认分支（无条件）
)
print(router.invoke({"q": "python装饰器是什么？"}))
```
调试技巧：管道中间插 `RunnableLambda(lambda x: print(x) or x)` 看中间结果。
|
|
工具的 docstring 就是给 LLM 看的描述（决定选参准确率）：
```python
from langchain_core.tools import tool
from pydantic import BaseModel, Field
@tool
def get_weather(city: str) -> str:
"""查询指定城市的实时天气。""" # docstring 必填，写清做什么
return f"{city}：晴，26°C"
class CalcInput(BaseModel):
expression: str = Field(description="数学表达式，如 3 * (4 + 5)")
@tool(args_schema=CalcInput) # args_schema 精确定义参数
def calculator(expression: str) -> str:
"""计算数学表达式。"""
return str(eval(expression))
```
|
|
工具调用本质 = LLM 输出结构化 tool_calls → 程序执行 → ToolMessage 回填 → 再生成：
```python
from langchain_core.messages import ToolMessage
llm_t = llm.bind_tools([get_weather, calculator])
ai_msg = llm_t.invoke("北京天气如何？顺便算 12.5*8")
print(ai_msg.tool_calls) # [{'name': 'get_weather', 'args': {'city': '北京'}, ...}]
tool_map = {t.name: t for t in [get_weather, calculator]}
msgs = [{"role": "user", "content": "..."}, ai_msg]
for tc in ai_msg.tool_calls:
result = tool_map[tc["name"]].invoke(tc["args"]) # 执行真实函数
msgs.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
final = llm_t.invoke(msgs) # LLM 基于工具结果生成自然语言回答
```
注意：一次可能返回**多个 tool_calls**（并行调用），每个都要回填对应 ToolMessage。
|
|
Agent = LLM + 工具 + 自动循环决策（ReAct），1.x 推荐 `langchain.agents.create_agent`：
```python
from langchain.agents import create_agent
agent = create_agent(
model=llm,
tools=[get_weather, calculator],
system_prompt="你是严谨的助手，能用工具解决的问题不要凭记忆回答。",
)
r = agent.invoke({"messages": [("user", "北京天气如何？换算成华氏(×1.8+32)")]},
{"recursion_limit": 10}) # 必须设上限，防循环烧钱
print(r["messages"][-1].content) # 自动多轮工具调用后的最终答案
```
⚠️ 版本要点：`langgraph.prebuilt.create_react_agent` 已废弃，迁移到 `langchain.agents.create_agent`（system 提示参数名 `system_prompt`）。存量代码里的 `AgentExecutor`（classic 层）仍可用但不再演进。
|
|
LCEL 走直线/分支；需要**循环、显式状态、断点续跑、人工介入**就上 LangGraph——把流程建成有向图：
```python
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
question: str
answer: str
logs: Annotated[list[str], add] # reducer：多节点写入时累加而非覆盖
b = StateGraph(State)
b.add_node("retrieve", retrieve_node) # 节点 = 函数，读写 state
b.add_node("generate", generate_node)
b.add_edge(START, "retrieve")
b.add_edge("retrieve", "generate")
b.add_edge("generate", END)
graph = b.compile()
print(graph.invoke({"question": "什么是RAG？", "answer": "", "logs": []}))
```
条件边实现分支/循环：`add_conditional_edges("grade", route_fn, {"ok": "generate", "retry": "rewrite"})`，把 rewrite 指回 retrieve 即成"检索→打分→重写→重检索"的自适应 RAG 循环。
|
|
checkpointer = 图级记忆（thread_id 区分会话，状态断点恢复）：
```python
from langgraph.checkpoint.memory import InMemorySaver
# 生产换 SqliteSaver / PostgresSaver
graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "user-001"}}
graph.invoke({"messages": [("user", "我叫李明")]}, config=cfg)
out = graph.invoke({"messages": [("user", "我叫什么？")]}, config=cfg) # 记得"李明"
```
interrupt 人工审批（敏感操作如删库/下单）：
```python
from langgraph.types import interrupt, Command
def approval(state):
d = interrupt({"question": "是否批准？"}) # 图在此挂起
return {"messages": [("system", "已批准" if d["approved"] else "已拒绝")]}
# 调用方：graph.invoke(Command(resume={"approved": True}), config=cfg) 注入决定续跑
```
|
|
| 需求 | 选择 |
|------|------|
| 直线管道 / 并行 / 简单分支 | LCEL（更轻） |
| 循环（重试/重写/多轮工具） | LangGraph |
| 显式共享状态、断点续跑 | LangGraph checkpointer |
| 人工介入审批 | LangGraph interrupt |
| 多 Agent 协作 | LangGraph subgraph |
子图即多智能体基元：
```python
main = StateGraph(State)
main.add_node("research", research_graph) # 编译好的子图直接当节点
main.add_node("write", write_graph)
main.add_edge(START, "research"); main.add_edge("research", "write"); main.add_edge("write", END)
```
|
|
LangServe 已废弃（与 1.x 不兼容），用纯 FastAPI 手写路由等价实现：
```python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="AI 服务")
chain = prompt | llm | StrOutputParser()
class JokeInput(BaseModel):
topic: str
@app.post("/joke/invoke")
async def joke_invoke(data: JokeInput):
return {"output": await chain.ainvoke({"topic": data.topic})}
# 终端启动：uvicorn joke_server:app --host 0.0.0.0 --port 8000
# 访问 http://localhost:8000/docs 自动 API 文档
```
⚠️ notebook 里**不要** `uvicorn.run`（`__name__ == "__main__"` 在 notebook 恒真，会永久阻塞内核）。
|
|
打字机效果的流式端点：
```python
from fastapi.responses import StreamingResponse
@app.post("/joke/stream")
async def joke_stream(data: JokeInput):
async def gen():
async for chunk in chain.astream({"topic": data.topic}):
yield f"data: {chunk}\n\n" # SSE 格式
yield "data: [DONE]\n\n"
return StreamingResponse(gen(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache",
"X-Accel-Buffering": "no"})
```
生产化三件套：`asyncio.wait_for` 超时 + `asyncio.Semaphore` 并发上限 + `/health` 健康检查；密钥走环境变量不进镜像；JWT 认证 + 限流保护端点。
|
|
追踪链/Agent 运行（零侵入，三个环境变量）：
```python
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_xxx" # smith.langchain.com 申请
os.environ["LANGCHAIN_PROJECT"] = "my-rag-demo"
# 之后所有链调用自动上报：prompt/输出/耗时/token/工具轨迹
```
评估闭环（改 prompt/换模型前跑一遍，用分数说话）：
```python
from langsmith import Client
from langsmith.evaluation import evaluate, LangChainStringEvaluator
# 建数据集(inputs/expected) → 被测函数 → criteria LLM 评委 → evaluate() 打分
correctness = LangChainStringEvaluator("criteria", config={"criteria": "correctness", "llm": llm})
results = evaluate(rag_answer, data="rag-qa-demo", evaluators=[correctness])
```
无 key 本地替代：自定义 `BaseCallbackHandler` 记录每步事件 + 关键词命中评估器。
|
|
0.x 教程代码在 1.x 环境的常见导入迁移：
| 旧（0.x） | 新（1.x） |
|-----------|-----------|
| langchain.memory | langchain_classic.memory |
| langchain.chains | langchain_classic.chains |
| langchain.prompts | langchain_core.prompts |
| langchain.schema | langchain_core.messages / documents |
| langchain.retrievers | langchain_classic.retrievers |
| langchain.vectorstores | langchain_community.vectorstores |
| langchain.document_loaders | langchain_community.document_loaders |
| langchain.text_splitter | langchain_text_splitters |
| langserve（已废弃） | FastAPI 手写路由 / LangGraph |
已物理删除：PipelinePromptTemplate（→LCEL 组合）、llm.predict（→invoke）、ElasticsearchRetriever.from_es_params（→构造函数 es_url 参数）。
|
|
2026-09 实测（lc_*_verify.ipynb 全量验证）高频坑：
1. **thinking 模型**：deepseek-v4-flash 下 `with_structured_output` 必须 `method="function_calling"` + `extra_body={"thinking": {"type": "disabled"}}`；`bind_tools`（auto）不受影响
2. **ToolNode 不能裸调用**：langgraph 1.2 起必须放 StateGraph 里（缺图 runtime）
3. **工具容错**：`@tool(handle_tool_error=)` 已失效；工具内抛 `ToolException` 才能被容错开关接住
4. **SelfQueryRetriever.from_llm 连环炸**：community 移除 DatabricksVectorSearch 所致，改手工构造（load_query_constructor_runnable + ChromaTranslator）
5. **Chroma**：集合名 ≥3 字符；持久化目录绑定 embedding 维度，换模型先删库；Windows 文件锁别运行中删目录
6. **环境降级**：ES/MCP 等可选依赖探测后优雅跳过，别阻断主流程
|
|
可选服务探测降级模式（教程/演示代码必备）：
```python
from elasticsearch import Elasticsearch
es = Elasticsearch(hosts=["http://localhost:9200"], request_timeout=5)
try:
es.info(); ES_OK = True
except Exception:
ES_OK = False
print("⚠️ 未检测到 ES，跳过相关演示") # 后续 ES cell 用 if ES_OK: 守卫
```
**完整学习路径回顾**：L1 会调通（模型/Prompt/解析）→ L2 备数据（加载/分割/向量库）→ L3 会组合（RAG/记忆/LCEL）→ L4 能上线（Agent/图编排/服务化/避坑）。
本页所有示例均在 `D:/wks/jupyter/lc_*_verify.ipynb` 实测通过（12 本 87 cell 全绿），对应知识库 zsku_lc_* 智能体家族（model/prompt/parser/memory/docvec/rag/lcel/tools/agent/graph/serve/eval/fix）。
|
LangChain 官方文档
LangGraph 文档
LangSmith 观测平台
MCP 协议