LangChain 使用方式与典型场景

使用方式与典型场景

示例一：用 LCEL 写一条 prompt | llm | parser 链

下面这段代码可以直接运行。它把「提示词模板 → 对话模型 → 文本解析」串成一条链，并顺带演示批量调用与流式输出两种方式。

import os
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

os.environ.setdefault("OPENAI_API_KEY", "sk-...")

# 1) 提示词模板：{topic} 是变量，运行时填充
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名技术编辑，回答要克制、准确，不要编造事实。"),
    ("human", "用三句话解释 {topic}，每句话不超过 40 字。"),
])

# 2) 模型：temperature 调低，减少随机发挥；超时与重试必须显式设置
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, timeout=30, max_retries=2)

# 3) 输出解析：把模型返回的 AIMessage 取成纯字符串
parser = StrOutputParser()

# 4) 用管道符组合，得到一条可调用链
chain = prompt | llm | parser

# 单次调用
print(chain.invoke({"topic": "向量数据库"}))

# 批量调用：一次传入多个输入，框架内部并发发起请求
print(chain.batch([{"topic": "向量数据库"}, {"topic": "提示词工程"}]))

# 流式输出：逐段打印，适合做成打字机效果
for chunk in chain.stream({"topic": "RAG 检索增强生成"}):
    print(chunk, end="", flush=True)

LCEL 的价值在于改动成本很低：想换成 ChatAnthropic 或本地模型，只改第 2 步那一行；想在返回前加个后处理，直接再 | RunnableLambda(清洗函数)；想改成异步，把 invoke 换成 ainvoke 就行，结构不用动。这也是它比手写调用函数更容易维护的地方。

示例二：VectorStore + Retriever 搭一个最小 RAG

RAG 的流程是「切分文档 → 向量化入库 → 按问题召回 → 拼进提示词让模型作答」。下面用一个内存版向量库写完整实现，换成 Chroma 或 pgvector 只需要替换向量库那一行。

from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1) 准备原始文档（真实场景用 TextLoader / PyPDFLoader 等加载器）
raw = """LangChain 是构建大模型应用的开发框架，提供提示词、模型、检索、工具等组件。
LCEL 是 LangChain 表达式语言，用管道符把组件串成链。
LangGraph 用状态图编排带分支和循环的多步流程，适合智能体。
RAG 是先检索再生成的做法，能缓解模型胡编的问题。"""

splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40)
chunks = splitter.split_text(raw)
docs = [Document(page_content=c, metadata={"source": "intro"}) for c in chunks]

# 2) 向量化并写入向量库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(docs, embeddings)

# 3) 转成检索器：每次召回 2 段
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 4) 提示词要求模型只依据上下文回答，并允许它说「不知道」
prompt = ChatPromptTemplate.from_messages([
    ("system", "只根据下面的上下文回答问题。上下文里没有的信息，直接回答「资料中没有提到」。\n\n上下文：\n{context}"),
    ("human", "{question}"),
])

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

print(chain.invoke("LCEL 是做什么的？"))

# 想核对召回了哪些片段，可以直接调检索器
for d in retriever.invoke("LangGraph 适合什么场景？"):
    print("-", d.page_content)

这段代码里有三个容易被忽略但很关键的点。第一，提示词里明确写了「资料中没有提到就直说」，没有这句约束，模型会用自己的知识补齐，RAG 就退化成了普通问答。第二，chunk_size 与 chunk_overlap 决定了召回质量的上限，建议拿真实文档试几组值做对比。第三，检索与生成是解耦的，你可以单独跑 retriever.invoke() 看召回了什么——这是排查「答案不对」时最快的一步：先确认是压根没召回，还是召回了但模型没用上。

示例三：用 LangGraph 编排带判断的流程

当流程里出现「先判断，再决定下一步走哪」时，链式写法会变得别扭，状态图更合适。它把每个步骤写成节点，节点之间用边连接，条件边负责分支。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    question: str
    category: str
    answer: str

def classify(state: State):
    # 真实项目里这里换成一次模型调用做意图分类
    q = state["question"]
    category = "技术" if ("报错" in q or "配置" in q) else "闲聊"
    return {"category": category}

def answer_tech(state: State):
    return {"answer": "已转技术文档检索：" + state["question"]}

def answer_small_talk(state: State):
    return {"answer": "你好，有什么可以帮你？"}

def route(state: State):
    return "tech" if state["category"] == "技术" else "chat"

graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("tech", answer_tech)
graph.add_node("chat", answer_small_talk)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route, {"tech": "tech", "chat": "chat"})
graph.add_edge("tech", END)
graph.add_edge("chat", END)

app = graph.compile()
print(app.invoke({"question": "这个报错怎么解决？"})["answer"])

状态图的好处是流程显式可见：有哪些节点、在哪里分叉、会不会循环，读一遍代码就能画出来。更复杂的场景还可以挂上检查点（checkpointer）做状态持久化，让流程中断后能从上次的位置继续——这也正是「先人工审核，通过后再继续」这类需求的基础。

适合与不适合的场景