LangChain 学习大纲

下面是一套“照着写就能学”的 LangChain 教程。每节只学一个重点；默认用 OpenAI，但 LangChain 的思想对其他模型也一样。

准备一次即可：

pip install -U langchain langchain-openai langchain-community langchain-text-splitters
pip install -U python-dotenv pydantic chromadb pypdf

新建 .env：

OPENAI_API_KEY=你的密钥

每个 Python 文件开头都放：

from dotenv import load_dotenv
load_dotenv()

不要把 .env 提交到 GitHub。

第 1 课：模型与 `invoke`

目标：能用 LangChain 调模型。

# lesson_01_model.py
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

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

response = model.invoke("用一句话解释 LangChain。")

print(response.content)
print(response.response_metadata)

记住：

ChatOpenAI 是模型对象。
invoke() 是单次调用。
返回的是 AIMessage，真正的文本在 .content。
temperature=0 更稳定，适合学习、抽取和 RAG。

练习：让模型把一句英文翻译成中文，并打印 response.content。

学会标准：不看教程，能传入一句话并取出模型回复。

官方参考：OpenAI 集成

第 2 课：Prompt 模板

目标：不把提示词拼接在 Python 字符串里。

# lesson_02_prompt.py
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

load_dotenv()

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是耐心的编程老师。回答控制在 100 字内。"),
    ("human", "请解释：{topic}")
])

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

messages = prompt.invoke({"topic": "什么是向量数据库？"})
response = model.invoke(messages)

print(response.content)

理解：

字典 {"topic": "..."}
        ↓
Prompt 模板
        ↓
消息列表（system + human）
        ↓
模型
        ↓
AIMessage

练习：增加 {level} 变量，例如“初学者”“有经验的后端工程师”，让不同水平的人得到不同解释。

学会标准：能区分 system prompt 和 human prompt，并可传入多个变量。

第 3 课：LCEL 链式写法

目标：学会 LangChain 最常见的 |。

# lesson_03_lcel.py
from dotenv import load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

load_dotenv()

prompt = ChatPromptTemplate.from_template(
    "用适合初学者的语言解释 {topic}，不超过三句话。"
)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

chain = prompt | model | parser

answer = chain.invoke({"topic": "RAG"})
print(answer)

这条链等于：

输入字典 → Prompt → Model → 把 AIMessage 转成 str

StrOutputParser() 的价值是：最后得到普通字符串，不必再写 .content。

练习：

for chunk in chain.stream({"topic": "Agent"}):
    print(chunk, end="", flush=True)

这就是流式输出。

学会标准：能解释 prompt | model | parser 每一段负责什么。

官方教程入口：LangChain 学习资源

第 4 课：批量、异步与 Runnable

目标：掌握链的执行模式。

# lesson_04_runnable.py
from dotenv import load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

load_dotenv()

prompt = ChatPromptTemplate.from_template("把下面句子翻译成英文：{text}")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | model | StrOutputParser()

# 1. 单次
print(chain.invoke({"text": "我在学习 LangChain。"}))

# 2. 批量
results = chain.batch([
    {"text": "你好"},
    {"text": "今天天气很好"},
])
print(results)

# 3. 插入普通 Python 函数
clean_text = RunnableLambda(lambda data: {"text": data["text"].strip()})
clean_chain = clean_text | chain

print(clean_chain.invoke({"text": "  LangChain 很有趣。  "}))

练习：写一个 RunnableLambda，把用户输入统一转成小写后再传给链。

学会标准：知道何时用：

invoke()：一个任务
batch()：一批互不依赖的任务
stream()：聊天界面、需要逐字展示
RunnableLambda：把自己的 Python 逻辑放入链条

第 5 课：结构化输出

目标：让模型返回可靠的数据对象，不要自己 json.loads() 猜格式。

# lesson_05_structured_output.py
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

load_dotenv()

class ArticleAnalysis(BaseModel):
    category: str = Field(description="文章类别，例如技术、财经、体育")
    sentiment: str = Field(description="情绪：正面、中性或负面")
    summary: str = Field(description="不超过30字的中文摘要")

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_model = model.with_structured_output(ArticleAnalysis)

result = structured_model.invoke(
    "某公司发布新一代芯片，性能提升40%，股价盘后上涨。"
)

print(result)
print(result.category)
print(result.summary)

练习：改成“简历信息提取器”，字段为姓名、技能列表、工作年限。

学会标准：模型返回的是 ArticleAnalysis 对象，而不是不稳定的文本。

第 6 课：RAG 的前半段——加载、切分、向量化

目标：把 PDF 变成可检索的文本块。

准备：把一个 PDF 放到项目目录，命名为 data.pdf。

# lesson_06_rag_index.py
from dotenv import load_dotenv
from langchain_community.document_loaders import PyPDFLoader
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma

load_dotenv()

# 1. 读取 PDF；每页通常成为一个 Document
pages = PyPDFLoader("data.pdf").load()

# 2. 切块；chunk_overlap 让上下文在边界处不断裂
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=80,
)
chunks = splitter.split_documents(pages)

# 3. 文本 → 向量，保存进本地 Chroma 数据库
vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db",
)

print(f"原始页数：{len(pages)}")
print(f"文本块数：{len(chunks)}")

额外安装：

pip install -U langchain-chroma

核心概念：

PDF
 ↓
Document（文本 + metadata，如页码）
 ↓
切成 chunks
 ↓
Embedding：每段文本变为数字向量
 ↓
Chroma：保存和搜索这些向量

练习：打印 chunks[0].page_content 和 chunks[0].metadata，确认页码元数据还在。

学会标准：你知道 Embedding 不是让模型“理解并回答”，而是把文本变成可按语义搜索的数据。

第 7 课：完成 RAG 问答

目标：只根据你的 PDF 回答，并返回来源页码。

# lesson_07_rag_chat.py
from dotenv import load_dotenv
from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

load_dotenv()

vector_store = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(
        model="text-embedding-3-small"
    ),
)

retriever = vector_store.as_retriever(search_kwargs={"k": 4})

def format_docs(docs: list[Document]) -> str:
    return "\n\n".join(
        f"[第 {doc.metadata.get('page', '?') + 1} 页]\n{doc.page_content}"
        for doc in docs
    )

prompt = ChatPromptTemplate.from_template("""
你是文档问答助手。只能根据下面的资料回答。
资料没有答案时，明确说“资料中没有找到答案”，不要自行编造。
回答末尾列出你参考的页码。

资料：
{context}

问题：{question}
""")

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

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": lambda question: question,
    }
    | prompt
    | model
    | StrOutputParser()
)

print(rag_chain.invoke("这份文档的核心结论是什么？"))

练习：

把 k=4 改成 k=2、k=8，比较答案质量。
问一个 PDF 中绝对没有的问题，确认它不会瞎答。
返回检索到的 docs，自己检查真正传给模型的上下文。

学会标准：能说清 RAG 的两个阶段：

索引阶段：加载 → 切分 → 向量库
问答阶段：检索 → 拼 Prompt → 模型生成

官方有对应的 PDF 语义搜索教程。

第 8 课：Tool Calling

目标：模型决定“要调用哪个工具”，程序执行工具。

# lesson_08_tools.py
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

load_dotenv()

@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积。"""
    return a * b

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。这里是示例数据，不是真实天气。"""
    weather = {
        "北京": "晴，18°C",
        "上海": "小雨，22°C",
    }
    return weather.get(city, "暂时没有该城市的天气数据。")

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools([multiply, get_weather])

response = model_with_tools.invoke("北京天气如何？顺便算 23 乘以 7。")

print(response.tool_calls)

这时不会自动运行工具。你应看到类似：

[
  {"name": "get_weather", "args": {"city": "北京"}},
  {"name": "multiply", "args": {"a": 23, "b": 7}}
]

练习：增加 convert_celsius_to_fahrenheit(celsius: float)。

学会标准：理解模型只“提议工具调用”；你的代码或 Agent 才真正执行函数。

第 9 课：Agent

目标：让框架完成“模型选工具 → 执行 → 将结果回传模型 → 给最终回答”的循环。

# lesson_09_agent.py
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

load_dotenv()

@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积。"""
    return a * b

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。此函数返回示例数据。"""
    data = {"北京": "晴，18°C", "上海": "小雨，22°C"}
    return data.get(city, "没有数据")

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

agent = create_agent(
    model=model,
    tools=[multiply, get_weather],
    system_prompt="你是简洁的生活助理。工具结果是可信来源。",
)

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "北京天气怎样？23 × 7 等于多少？"}
    ]
})

print(result["messages"][-1].content)

练习：

问“什么是 LangChain？”：观察 Agent 不调用工具。
问“上海天气和 15×8”：观察它调用多个工具。
用 get_weather 返回错误信息，要求 Agent 对用户坦诚说明。

学会标准：能回答：固定流程为什么优先 LCEL？因为它更可控、易测试、通常更便宜；需求确实需要模型决定步骤和工具时，才用 Agent。

官方学习页里有 RAG Agent、SQL Agent 等项目教程。

第 10 课：LangGraph

目标：理解什么时候不该继续堆 Agent，而要用工作流图。

先只学这三个概念：

State：整个任务共享的数据
Node：一个处理步骤，例如“检索”“生成”“人工审核”
Edge：从一个步骤走向下一个步骤的规则

适合 LangGraph 的情况：

回答前必须人工审核
失败后要重试或换策略
有循环，例如“检索不足 → 改写问题 → 再检索”
多个角色有明确分工
任务中断后要恢复

你的首个练习建议：把第 7 课改成流程图：

用户问题
  ↓
检索文档
  ↓
判断资料是否足够
  ├─ 足够 → 生成带引用的回答
  └─ 不足 → 回答“资料不足”

不要一开始做多 Agent；先把单 Agent + RAG 做扎实。官方也提供 LangGraph 自定义 RAG Agent 教程。

建议你真正执行的顺序：

第 1–3 课：一天
第 4–5 课：一天
第 6–7 课：两天
第 8–9 课：两天
第 10 课：完成前面项目后再开始

完成第 7 课后，你就已经能独立做有价值的 LangChain 项目。

第 1 课：模型与 invoke