上一章我们学会了用 FastAPI 把数据变成服务。但现在的 AI 时代，光能调用 API 还不够——我们需要让 API **更聪明**。
这就是本期的主题：**AI 实战——用 LangChain 调用大模型 + RAG 知识库**。
我们要做的不是简单地调 ChatGPT API 聊天，而是：
- 用 **LangChain** 框架把 AI 变成可复用的工具链
- 搭建 **RAG（检索增强生成）** 系统，让 AI 能回答你自己的文档问题
- 结合前面的 FastAPI，做一个真正的 **AI 服务项目**
>**前置知识**：需要掌握 Episode 07 的 FastAPI 基础和 Episode 06 的数据处理基础。
---
## 8.1 为什么要学 LangChain？
### 8.1.1 直接调 API vs 用框架
如果你只是想调用 ChatGPT，确实可以直接发 HTTP 请求：
```python
import requests
response = requests.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": "Bearer YOUR_KEY"},
json={
"model": "gpt-4",
"messages": [{"role": "user", "content": "你好"}]
}
)
```
但这有问题：
- 你需要自己处理 token 管理、重试、流式输出
- 想让 AI 先查数据库再回答？没有现成的方法
- 想让 AI 读取你的 PDF 文档并回答问题？从头写很麻烦
**LangChain 解决了这些问题**。它是一个"AI 应用开发框架"，把常见的 AI 任务封装成了标准化的组件。
```
LangChain 的三大核心价值：
1. Chain（链式调用）——把多个 AI 步骤串在一起
2. Memory（记忆）——记住对话上下文
3. Integrations（集成）——无缝对接各种数据源和模型
```
### 8.1.2 支持的模型
LangChain 支持几乎所有主流大模型：
| 提供商 | 模型 | API 基础 URL |
|--------|------|-------------|
| OpenAI | GPT-4o, GPT-3.5 | `https://api.openai.com/v1` |
| Anthropic | Claude 3 | `https://api.anthropic.com` |
| 国内厂商 | 通义千问、文心一言、Kimi | 各自的 API |
| 开源 | Llama 3、Mistral | 本地部署 |
>**教程中使用 OpenAI 格式**（几乎所有国内模型都兼容 OpenAI 格式），替换 `base_url` 就能切换模型。
---
## 8.2 第一个 LangChain 应用
### 8.2.1 安装
```bash
pipinstalllangchain-openailangchain_core
```
### 8.2.2 最简单的对话
```python
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
# 1. 选择模型——OpenAI 格式（兼容 OpenAI、通义千问、Ollama 等）
llm = ChatOpenAI(
model="gpt-4o-mini", # 也可以换成 qwen-turbo, claude-3-haiku 等
temperature=0.7, # 创造性程度：0=确定性强，1=创意强
base_url="https://api.openai.com/v1", # 改成其他模型的服务地址
api_key="your-api-key"
)
# 2. 发送消息
response = llm.invoke([
SystemMessage(content="你是一个乐于助人的 Python 编程助教。"),
HumanMessage(content="请用一句话解释什么是递归函数。")
])
print(response.content)
# 递归函数就是在函数体内调用自身的一个函数，通常需要一个终止条件来防止无限循环。
```
### 8.2.3 流式输出——像 ChatGPT 一样逐字显示
```python
from langchain_core.messages import HumanMessage
for chunk in llm.stream([
HumanMessage(content="给我讲一个关于 Python 的小笑话。")
]):
print(chunk.content, end="", flush=True)
# 为什么 Python 程序员总是喜欢写代码？因为他们不喜欢 debug...哦等等，他们其实也很讨厌 debug 😄
```
---
## 8.3 Prompt Template：让 AI 输出格式可控
### 8.3.1 什么是 Prompt Template？
直接把用户的话丢给 AI 效果很差。好的 Prompt 是 AI 应用的"灵魂"。
```python
from langchain_core.prompts import ChatPromptTemplate
# 1. 手动写 Prompt（推荐做法：用模板）
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的 {language} 编程老师，擅长用通俗的语言教学。"),
("human", "请用 {level} 水平的语言解释：{topic}"),
])
# 2. 渲染模板
chain = prompt | llm
response = chain.invoke({
"language": "中文",
"level": "零基础",
"topic": "什么是列表推导式"
})
print(response.content)
# 列表推导式就像是一个"魔法公式"，你可以在一行代码里完成两件事：
# 1. 遍历一堆东西（比如一个列表）
# 2. 对每个东西做处理，然后把结果收集到一个新列表里
```
### 8.3.2 结构化输出：让 AI 返回 JSON
```python
from langchain_core.output_parsers import JsonOutputParser
# 定义期望的输出格式
parser = JsonOutputParser(pydantic_object=dict)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个数据分析报告生成器。请严格按照以下 JSON 格式输出：\n{format_instructions}"),
("human", "请分析以下数据并给出摘要：{data}"),
])
# 组合链
chain = prompt | llm | parser
result = chain.invoke({
"format_instructions": parser.get_format_instructions(),
"data": "2026年上半年电商数据：总销售额¥125万，同比增长35%。服装类目增长最快(+58%)。"
})
print(result)
# {
# "total_sales": 1250000,
# "growth_rate": "35%",
# "fastest_growing_category": "服装",
# "category_growth": {"服装": "+58%"},
# "summary": "2026年上半年电商业务表现强劲，服装类目是主要增长引擎。"
# }
```
---
## 8.4 工具链（Tools）：让 AI 能"动手干活"
LangChain 最重要的概念之一：**给 AI 提供工具**。AI 不仅能回答问题，还能**执行具体的任务**。
### 8.4.1 定义工具
```python
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
@tool
defsearch_weather(city: str) -> str:
"""查询指定城市的天气情况。"""
# 在实际应用中这里会调天气 API
weather_data = {
"北京": "晴天，15°C ~ 25°C",
"上海": "多云，18°C ~ 26°C",
"深圳": "小雨，22°C ~ 30°C"
}
return weather_data.get(city, f"暂未收录 {city} 的天气数据，请使用天气 API 查询。")
@tool
defcalculate(expression: str) -> str:
"""计算数学表达式。"""
try:
# ⚠️ 生产环境不要用 eval，这里仅作演示
result = eval(expression, {"__builtins__": {}}, {})
returnstr(round(result, 4))
exceptExceptionas e:
returnf"计算出错: {e}"
# 创建工具列表
tools = [search_weather, calculate]
# 创建 prompt
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个乐于助人的 AI 助手。当用户问的问题需要你使用工具时，请调用对应的工具。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
# 创建 Agent
llm_with_tools = llm.bind_tools(tools)
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 测试
print(agent_executor.invoke({"input": "北京的天气怎么样？"})["output"])
# 工具调用 → search_weather("北京") → "晴天，15°C ~ 25°C"
print(agent_executor.invoke({"input": "1234 * 5678 等于多少？"})["output"])
# 工具调用 → calculate("1234 * 5678") → "7006652"
```
### 8.4.2 实战：数据分析助手
```python
import pandas as pd
# 模拟一个销售数据集
sales_df = pd.DataFrame({
"月份": ["1月", "2月", "3月", "4月", "5月", "6月"],
"线上销售额": [120, 135, 110, 142, 150, 155], # 万元
"线下销售额": [80, 75, 90, 85, 95, 100],
"活跃用户数": [5000, 5200, 5100, 5500, 5800, 6200]
})
@tool
defanalyze_sales(query: str) -> str:
"""分析销售数据，支持查询某个月的数据、总计、趋势等。"""
return sales_df.query(query).to_string(index=False) if query else sales_df.to_string(index=False)
@tool
defget_sales_summary() -> str:
"""获取销售数据摘要统计。"""
summary = sales_df[["线上销售额", "线下销售额", "活跃用户数"]].describe()
return summary.to_string()
sales_tools = [analyze_sales, get_sales_summary]
# 这个 AI 现在可以直接用 Pandas 帮你分析数据了！
```
---
## 8.5 RAG（检索增强生成）：让 AI 读你的文档
这是本期最重要的部分。**RAG** 是目前 AI 落地最热门的方向之一。
### 8.5.1 什么是 RAG？
```
传统 LLM 的问题：
- 训练数据截止于某个日期
- 不知道你的私有数据（公司内部文档、个人笔记）
- 容易"幻觉"——一本正经地胡说八道
RAG 的解决方案：
1. 你把文档喂给系统
2. 用户提问
3. 系统先从文档中找到相关内容
4. 把这些内容和用户问题一起发给 AI
5. AI 基于找到的内容回答问题
结果：回答准确、有据可依、不容易幻觉。
```
### 8.5.2 搭建 RAG 系统
```bash
pipinstalllangchain-communitychromadbtiktoken
```
```python
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
import os
# ============ 第一步：准备文档 ============
# 假设你有一堆 Markdown 文档放在 docs/ 目录下
docs_dir = "docs"
documents = []
for filename in os.listdir(docs_dir):
if filename.endswith(".md"):
loader = TextLoader(os.path.join(docs_dir, filename))
documents.extend(loader.load())
print(f"加载了 {len(documents)} 份文档")
# ============ 第二步：切分文档（Chunking） ============
# AI 一次只能处理有限长度的文本，所以要切成小块
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每个 chunk 约 500 字符
chunk_overlap=50, # 相邻 chunk 重叠 50 字符（避免边界丢失信息）
)
chunks = splitter.split_documents(documents)
print(f"切分成 {len(chunks)} 个片段")
# ============ 第三步：向量化存储（Embedding + Vector Store） ============
# 把每个文本片段转换成向量（数字表示语义）
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
# ChromaDB：轻量级向量数据库，存在本地
vectorstore = Chroma.from_documents(documents=chunks, embedding=embeddings)
# ============ 第四步：构建 RAG 链 ============
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# retriever：从向量数据库中检索最相关的文档片段
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3} # 返回最相关的 3 个片段
)
# RAG 模板
rag_template = """基于以下参考信息回答问题。如果参考信息不足以回答，请说"我不确定"。
参考信息：
{context}
用户问题：{question}
请给出简洁准确的答案：
"""
prompt = ChatPromptTemplate.from_messages([
("system", rag_template),
("human", "{question}"),
])
# 组合 RAG 链
defformat_docs(docs):
return"\n\n---\n\n".join([d.page_content for d in docs])
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# ============ 第五步：测试 ============
# 假设你的 docs/ 目录下有 Python 基础教程、数据分析笔记等文档
answer = rag_chain.invoke("Python 中的 Pandas 是什么？怎么用 groupby？")
print(answer)
# 基于你的文档内容生成回答...
```
### 8.5.3 RAG 质量优化技巧
```python
# 1. 调整 chunk 大小——太小有语境缺失，太大检索不精准
# 推荐：500-1000 字符（中文），200-400 token（英文）
splitter = RecursiveCharacterTextSplitter(
chunk_size=800,
chunk_overlap=100,
separators=["\n\n", "\n", "。", ".", "!"] # 优先在段落/句子边界切分
)
# 2. 使用混合检索——关键词 + 语义
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import EmbeddingsFilter
# 先用关键词匹配候选集，再用语义排序
# （需要安装 elasticsearch 或 jieba）
# 3. 重新排序（Re-ranking）——检索后用更精细的模型重排
# pip install langchain-cohere
# 检索到 20 个 → Reranker 排序 → 取前 5 个
# 4. 元数据过滤——按标签/分类/时间筛
# vectorstore.as_retriever(search_kwargs={"k": 5, "filter": {"source": "python_tutorial"}})
```
---
## 8.6 实战：做一个"智能知识助手"Web 应用
把前面学到的所有东西组合起来：**LangChain RAG + FastAPI = 一个完整的 AI 知识服务平台**。
```python
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import shutil
app = FastAPI(title="AI 知识助手")
classQuestionRequest(BaseModel):
question: str
top_k: int = 3
# 初始化 RAG 系统
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
vectorstore = ChromaCollection.persisted(
collection_name="knowledge_base",
embedding_function=embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
@staticmethod
defbuild_rag_chain():
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个智能知识助手。请基于以下参考信息回答用户的问题。
如果参考信息不足以回答，请诚实告知用户，并说明你需要什么资料。
参考信息：
{context}
---
用户问题：{question}
"""),
])
return prompt | llm
rag_chain = build_rag_chain()
defformat_docs(docs):
return"\n\n=== 参考文档 ===\n\n".join(
[f"[文档{i+1}]\n{d.page_content}"for i, d inenumerate(docs)]
)
@app.post("/ask")
defask_question(req: QuestionRequest):
"""提问接口"""
docs = retriever.invoke(req.question)
context = format_docs(docs)
response = rag_chain.invoke({
"context": context,
"question": req.question
})
return {
"question": req.question,
"answer": response.content,
"sources": [d.metadata.get("source", "未知") for d in docs]
}
@app.post("/add-knowledge")
defadd_knowledge(filename: str, content: str):
"""添加新知识"""
# 保存到本地后重新构建向量库
withopen(f"knowledge/{filename}.md", "w", encoding="utf-8") as f:
f.write(content)
return {"status": "ok", "message": f"知识 '{filename}' 已保存"}
@app.get("/health")
defhealth():
return {"status": "running", "models": {"llm": "gpt-4o-mini", "embedding": "text-embedding-ada-002"}}
```
启动：
```bash
uvicornmain:app--reload
```
访问 `http://127.0.0.1:8000/docs` 测试你的 AI 知识助手！
---
## 8.7 LangChain 常用组件速查表
| 组件 | 用途 | 核心类 |
|------|------|--------|
| LLM 调用 | 调用大模型 | `ChatOpenAI` |
| Embedding | 文本向量化 | `OpenAIEmbeddings` |
| Prompt 模板 | 结构化提示词 | `ChatPromptTemplate` |
| 输出解析器 | 让 AI 输出 JSON | `JsonOutputParser` |
| 工具（Tools） | 让 AI 执行操作 | `@tool` 装饰器 |
| Agent | AI + 工具的自主决策 | `AgentExecutor` |
| Vector Store | 向量数据库 | `Chroma`, `FAISS` |
| Retriever | 语义检索 | `vectorstore.as_retriever()` |
| Chain | 链式处理 | `prompt | llm | parser` |
---
## 8.8 练习题
### 练习 1：构建一个"代码调试助手"
用 LangChain 做一个工具：
- 用户提供一段报错的代码
- 助手先分析代码逻辑
- 然后用 `execute_code` 工具试运行
- 最后给出修改建议
```python
# 参考思路
@tool
defexecute_code(code: str) -> str:
"""执行一段 Python 代码并返回结果。"""
try:
exec(code, {"__builtins__": {}}, {})
return"代码执行成功"
exceptExceptionas e:
returnf"执行失败: {e}"
```
### 练习 2：改进 RAG 检索质量
找一份技术文档（如 30 页的 PDF），搭建 RAG 系统：
1. 尝试不同的 chunk_size（100/500/1000），比较检索效果
2. 添加 source 元数据，确保回复中标注引用来源
3. 实现一个 FAQ 模式：常见问答用关键词匹配，开放问题用语义检索
### 练习 3：构建多轮对话记忆
```python
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# 这个对话链会记住之前聊过的内容
memory = ConversationBufferMemory()
conversation = ConversationChain(llm=llm, memory=memory)
conversation.predict(input="我叫小明，今年25岁。")
conversation.predict(input="我叫什么？") # 应该回答"小明"
```
### 练习 4：把数据分析结果喂给 AI 生成报告
结合 Episode 06 的 Pandas 分析，把统计结果送给 AI：
```python
import pandas as pd
# 分析电商数据
df = pd.read_csv("sales.csv")
stats = df.describe().to_dict()
# 喂给 AI
response = llm.invoke([
HumanMessage(content=f"""以下是某电商平台6月份的销售统计数据：
{stats}
请基于这些数据生成一份简短的业务分析报告（200字以内）。
重点关注：增长机会、风险点和行动建议。""")
])
```
### 练习 5：做一个语音问答助手
```python
# 语音输入 → 文字识别 → RAG 检索 → AI 回答 → 语音输出
# 工具库：speech_recognition + pyttsx3
```
---
## 8.9 知识点小结
| 知识点 | 关键词 |
|--------|--------|
| LLM 调用 | ChatOpenAI / temperature / 流式输出 |
| Prompt Engineering | ChatPromptTemplate / 结构化输出 / JsonOutputParser |
| Agent 与 Tools | @tool / AgentExecutor / 自主决策 |
| RAG 核心流程 | 文档加载 → 切分 → 向量化 → 检索 → 生成 |
| 向量数据库 | Chroma / FAISS / 语义检索 |
| 链式处理 | RunnablePassthrough / 管道组合 |
---
## 8.10 工具链总结：AI 赋能一切
```
第05期 第06期 第07期 第08期
爬虫 → 数据 数据分析 → 洞察 FastAPI → 服务 LangChain + RAG → 智能
↓
用 AI 理解和回答一切
```
你现在拥有了一条完整的技术链：
1.**获取数据**（爬虫）
2.**分析数据**（Pandas）
3.**提供服务**（FastAPI）
4.**智能理解**（LangChain + RAG）
下一期（Episode 09），我们将跨入**前端世界**——学习如何用 React/Vue 做一个漂亮的用户界面，让你的 API 和 AI 服务能力被普通人直接使用！
准备好你的 AI Key，我们下期见！🐍🤖✨
10个月宝宝每天需要喝多少奶粉？