Skip to content

LangChain 从入门到实战精通

导航目录


一、LangChain 是什么

1.1 核心说明

LangChain 是一个用于构建大模型应用的开发框架,它把 Prompt、模型调用、工具调用、检索、记忆、工作流编排等能力标准化,帮助你快速搭建可维护的 AI 应用。

典型应用场景:

  • 智能对话机器人
  • 知识库问答(RAG)
  • 文档分析与摘要
  • 多工具联动自动化
  • API 化对外服务

1.2 最小可运行示例

python
# 基础链搭建代码示例
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

os.environ["OPENAI_API_KEY"] = "你的API_KEY"

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template("请用一句话解释:{topic}")
chain = prompt | model

resp = chain.invoke({"topic": "什么是LangChain"})
print(resp.content)
# 效果说明:返回一个可读的解释文本

1.3 关键注意点

  • LangChain 不替代模型,它是模型应用编排层。
  • 先学会 Prompt -> Model -> Output 三件套,再上 RAG/Agent。

二、为什么项目里会引入 LangChain

2.1 核心价值

  • 统一接口:多模型(OpenAI/Anthropic/本地模型)切换成本低
  • 组件化:Prompt、Retriever、Tool、Memory 可插拔
  • 可观测:支持 tracing、日志、错误恢复
  • 工程化友好:适合 API 服务化与持续迭代

2.2 基础开发流程(完整)

  1. 环境搭建
  2. 安装依赖
  3. 配置 API Key / Base URL
  4. 初始化模型
  5. 组装 Prompt/Chain
  6. 联调与测试
  7. 增加检索、工具、记忆
  8. 服务化与监控

2.3 完整入门流程代码

python
# 基础开发流程代码示例
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1) 配置环境变量
os.environ["OPENAI_API_KEY"] = "你的API_KEY"

# 2) 初始化模型
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.2,   # 输出稳定性更高
    max_tokens=300,    # 控制最大输出长度
    timeout=30,        # 网络超时
)

# 3) Prompt 模板
prompt = ChatPromptTemplate.from_template(
    "你是资深工程师。请用3点解释 {topic},每点不超过20字。"
)

# 4) 组装链
chain = prompt | llm | StrOutputParser()

# 5) 执行测试
result = chain.invoke({"topic": "LangChain的核心价值"})
print(result)
# 效果说明:输出格式稳定、可直接展示到前端

三、核心包与安装建议

3.1 Python 版本与核心依赖

  • Python:建议 3.10+
  • 核心包:
    • langchain
    • langchain-core
    • langchain-openai
    • langchain-community
  • RAG 常用:
    • chromadb
    • faiss-cpu
    • pypdf

3.2 安装命令(pip + 镜像)

bash
# 标准安装
pip install -U langchain langchain-core langchain-openai langchain-community
pip install -U chromadb faiss-cpu pypdf fastapi uvicorn python-dotenv

# 国内镜像(避免下载失败)
pip install -U langchain langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple

3.3 标准导入与验证

python
# 导入验证代码示例
import langchain
import langchain_core
import langchain_openai

print("langchain:", langchain.__version__)
print("langchain_core:", langchain_core.__version__)
print("langchain_openai:", langchain_openai.__version__)
# 效果说明:能输出版本号即环境可用

3.4 新手常见误区与避坑

  • 依赖冲突:统一升级到最新版,必要时新建虚拟环境
  • API Key 未生效:确认 OPENAI_API_KEY 在当前进程可见
  • 网络超时:设置 timeout,并加重试机制

四、模型、消息与 Prompt 模板

4.1 核心概念速记

  • LLM / ChatModel:负责生成内容
  • PromptTemplate:参数化提示词
  • Message:system/human/ai 消息结构
  • Chain:把多个步骤串起来
  • Tool:可调用的函数能力

4.2 PromptTemplate 基础与动态模板

python
# PromptTemplate 基础代码示例
from langchain_core.prompts import PromptTemplate

tpl = PromptTemplate.from_template(
    "请将下面内容改写为更专业语气:\n{text}"
)
print(tpl.format(text="这个功能挺好用的。"))
# 效果说明:输出渲染后的 prompt 文本

4.3 条件模板(通过分支实现)

python
# 条件模板代码示例
from langchain_core.prompts import ChatPromptTemplate

def build_prompt(level: str):
    if level == "beginner":
        return ChatPromptTemplate.from_template("请用新手能懂的方式解释:{topic}")
    return ChatPromptTemplate.from_template("请从工程实践角度深入解释:{topic}")

prompt = build_prompt("beginner")
print(prompt.format_messages(topic="向量数据库"))
# 效果说明:根据 level 返回不同复杂度提示词

4.4 批量模板生成

python
# 批量模板调用示例
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template("一句话总结:{topic}")
chain = prompt | llm

inputs = [{"topic": "RAG"}, {"topic": "Agent"}, {"topic": "Tool Calling"}]
responses = chain.batch(inputs)
for r in responses:
    print(r.content)
# 效果说明:一次请求批量处理多个输入,效率更高

4.5 参数说明(常用)

  • model:模型名称,如 gpt-4o-mini
  • temperature:随机性,0~2,越低越稳定
  • max_tokens:单次最大输出长度
  • timeout:请求超时秒数

五、结构化输出与解析器

5.1 为什么要结构化输出

在生产中,返回 JSON 比自然语言更易存储、校验、接口传输。

5.2 Pydantic 结构化输出

python
# 结构化输出代码示例
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class Plan(BaseModel):
    title: str = Field(description="任务标题")
    steps: List[str] = Field(description="执行步骤")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(Plan)
prompt = ChatPromptTemplate.from_template("给我一个学习{topic}的3步计划")

chain = prompt | structured_llm
data = chain.invoke({"topic": "LangChain"})
print(data)
# 效果说明:得到 Plan 对象而不是纯文本

5.3 常见问题

  • 输出字段缺失:在提示词中补充字段约束
  • 解析失败:降低温度并强化格式要求

六、Runnable 与 LCEL

6.1 核心说明

LCEL(LangChain Expression Language)通过 | 把组件串联:
prompt | model | parser

6.2 Runnable 组合示例

python
# Runnable 与 LCEL 代码示例
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("把这句话翻译成英文:{text}")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

chain = prompt | llm | parser
print(chain.invoke({"text": "今天心情很好"}))
# 效果说明:输出纯字符串结果

6.3 并行与映射

python
# 并行执行示例
from langchain_core.runnables import RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

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

zh = ChatPromptTemplate.from_template("翻译为中文:{text}") | llm | parser
en = ChatPromptTemplate.from_template("翻译为英文:{text}") | llm | parser

parallel = RunnableParallel(to_zh=zh, to_en=en)
print(parallel.invoke({"text": "LangChain improves LLM engineering."}))
# 效果说明:一次输入,同时得到两个方向的结果

七、链式调用与组合模式

7.1 LLMChain / SequentialChain 说明

  • LLMChain:单步骤处理
  • SimpleSequentialChain:多个步骤串行
  • SequentialChain:多输入多输出串行

7.2 对比代码示例

python
# Chain 对比代码示例
from langchain_openai import OpenAI
from langchain.chains import LLMChain, SimpleSequentialChain
from langchain.prompts import PromptTemplate

llm = OpenAI(temperature=0)

p1 = PromptTemplate(input_variables=["topic"], template="给出{topic}的一个标题")
c1 = LLMChain(llm=llm, prompt=p1)

p2 = PromptTemplate(input_variables=["title"], template="根据标题写一句摘要:{title}")
c2 = LLMChain(llm=llm, prompt=p2)

pipeline = SimpleSequentialChain(chains=[c1, c2], verbose=True)
print(pipeline.run("LangChain学习路线"))
# 效果说明:先生成标题,再生成摘要

7.3 复杂 Chain(分支/循环/多模型)

python
# 分支 + 多模型联动思路示例
# 注意:示例展示架构思想,可在 LCEL 中通过 RunnableBranch/自定义函数实现
"""
1) 输入问题
2) 分类器判断:技术类/业务类
3) 技术类走模型A,业务类走模型B
4) 统一后处理并输出
"""
# 效果说明:复杂业务中可按任务类型切模型和处理路径

八、向量、Embedding 与 Retriever

8.1 核心说明

RAG 基础链路:文档 -> 向量化 -> 存储 -> 检索 -> 生成

8.2 Chroma 向量库示例

python
# Embedding + Chroma 代码示例
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

emb = OpenAIEmbeddings(model="text-embedding-3-small")
docs = [
    Document(page_content="LangChain可以构建Agent系统", metadata={"source": "a"}),
    Document(page_content="RAG用于外部知识增强", metadata={"source": "b"}),
]
vs = Chroma.from_documents(documents=docs, embedding=emb, collection_name="demo_docs")

hits = vs.similarity_search("什么是RAG", k=2)
for h in hits:
    print(h.page_content, h.metadata)
# 效果说明:返回最相近的文档片段

8.3 FAISS 示例

python
# Embedding + FAISS 代码示例
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

emb = OpenAIEmbeddings(model="text-embedding-3-small")
texts = ["苹果是水果", "汽车用于出行", "数据库用于存储数据"]
vs = FAISS.from_texts(texts, emb)
print(vs.similarity_search("什么用于存储信息", k=1)[0].page_content)
# 效果说明:检索到“数据库用于存储数据”

8.4 参数说明

  • k:召回文档数量,常用 3~8
  • score_threshold:相似度阈值(按库支持)
  • chunk_size/chunk_overlap:影响检索粒度与上下文完整性

九、RAG 的完整实现思路

9.1 全流程

  1. 文档加载(PDF/TXT/Word)
  2. 文本切分
  3. 向量化入库
  4. Retriever 检索
  5. 组装问答链

9.2 可直接复用代码

python
# 知识库问答完整代码示例
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = "你的API_KEY"

# 1) 文档加载(按你的文件类型选择)
docs = TextLoader("kb.txt", encoding="utf-8").load()
# docs = PyPDFLoader("kb.pdf").load()

# 2) 切分
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=80)
chunks = splitter.split_documents(docs)

# 3) 向量存储
emb = OpenAIEmbeddings(model="text-embedding-3-small")
vs = Chroma.from_documents(chunks, emb, collection_name="kb_collection")
retriever = vs.as_retriever(search_kwargs={"k": 4})

# 4) 问答链
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template(
    "你是知识库问答助手。仅基于上下文回答。\n\n上下文:\n{context}\n\n问题:{question}"
)
parser = StrOutputParser()

def ask(question: str) -> str:
    docs = retriever.invoke(question)
    context = "\n\n".join(d.page_content for d in docs)
    chain = prompt | llm | parser
    return chain.invoke({"context": context, "question": question})

print(ask("这份知识库主要讲了什么?"))
# 效果说明:回答内容受知识库约束,幻觉显著减少

9.3 注意事项

  • 检索不准先调 chunk_size/k
  • 对高精度场景加 rerank
  • 答案要“带来源”便于追溯

十、工具调用、Tool Calling 与 MCP

10.1 内置 Tool 与自定义 Tool

python
# 自定义 Tool 代码示例
from langchain_core.tools import tool

@tool
def add_numbers(a: int, b: int) -> int:
    """计算两个整数之和"""
    return a + b

print(add_numbers.invoke({"a": 3, "b": 5}))
# 效果说明:返回 8

10.2 Tool 与 Chain 联动逻辑

  • Chain 负责语言理解
  • Tool 负责外部动作(查库/查天气/调用业务系统)
  • Agent 负责在二者之间“决策调度”

10.3 工具权限控制建议

  • 只暴露必要工具
  • 为工具输入做 schema 校验
  • 对高风险工具(写操作)增加白名单和审批

10.4 MCP 说明

MCP 可把外部系统能力标准化为可调用工具资源,LangChain 可通过工具层接入,形成统一调度链路。


十一、Agent、ReAct 与多 Agent 协作

11.1 Agent 类型与适用场景

  • 基础 Agent:简单问答 + 工具调用
  • ReAct Agent:边思考边行动,适合多步推理
  • Self-Ask:先拆问题再求解,适合复杂问答
  • Plan-and-Execute:先规划后执行,适合流程任务

11.2 ReAct Agent 示例

python
# ReAct Agent 代码示例
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """计算乘法"""
    return a * b

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [multiply]

prompt = PromptTemplate.from_template(
    """你可以使用工具解决问题。
问题: {input}
可用工具: {tools}
工具名称: {tool_names}
{agent_scratchpad}"""
)

agent = create_react_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
print(executor.invoke({"input": "7乘以9是多少?"}))
# 效果说明:Agent 会自动调用 multiply 工具

11.3 Plan-and-Execute 思路

python
# Plan-and-Execute 架构示例(伪代码)
"""
planner: 负责把目标拆成子任务
executor: 逐个执行子任务(可调用多个工具)
supervisor: 汇总结果并给出最终答复
"""
# 注意事项:计划要可观测、可中断、可重试

11.4 多 Agent 协作建议

  • 角色分工明确:规划/执行/审查
  • 共享上下文最小化:避免上下文污染
  • 每个 Agent 限制工具权限

十二、记忆、状态与 LangGraph 的边界

12.1 多轮对话记忆(机器人场景)

python
# 多轮对话记忆代码示例(简化)
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
history = [SystemMessage(content="你是专业客服助手。")]

def chat(user_text: str) -> str:
    history.append(HumanMessage(content=user_text))
    resp = llm.invoke(history)
    history.append(AIMessage(content=resp.content))
    return resp.content

print(chat("我想了解你们的退款规则"))
print(chat("请结合我刚才的问题,再给一个简短版本"))
# 效果说明:第二轮可引用第一轮上下文

12.2 LangChain 与 LangGraph 边界

  • LangChain:线性/轻量编排快
  • LangGraph:复杂状态机、多分支回退、长流程更强

12.3 何时升级到 LangGraph

  • 需要强状态控制
  • 需要人工中断与恢复
  • 需要多 Agent 工作流编排

十三、生产实践与常见坑

13.1 错误处理、日志、重试(可复用)

python
# 稳定性增强代码示例
import time
import logging
from langchain_openai import ChatOpenAI

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("llm_service")

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

def safe_invoke(messages, retry=3):
    last_err = None
    for i in range(retry):
        try:
            return llm.invoke(messages)
        except Exception as e:
            last_err = e
            logger.warning("调用失败,第%s次重试,错误=%s", i + 1, str(e))
            time.sleep(2 ** i)
    raise RuntimeError(f"模型调用失败: {last_err}")

13.2 FastAPI 封装(API 接口开发)

python
# FastAPI 接口实战代码示例
import os
from fastapi import FastAPI
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = "你的API_KEY"

app = FastAPI(title="LangChain API Demo")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
prompt = ChatPromptTemplate.from_template("请回答问题:{question}")
chain = prompt | llm | StrOutputParser()

class AskReq(BaseModel):
    question: str

@app.post("/ask")
def ask(req: AskReq):
    answer = chain.invoke({"question": req.question})
    return {"answer": answer}

# 运行方式:
# uvicorn app:app --host 0.0.0.0 --port 8000 --reload

13.3 本地化部署(离线可控)

python
# 本地模型接入示例(以本地OpenAI兼容网关为例)
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="qwen2.5-7b-instruct",
    api_key="EMPTY",
    base_url="http://127.0.0.1:8000/v1",  # 本地兼容OpenAI接口服务
    temperature=0.2,
)
print(llm.invoke("请用一句话介绍你自己").content)
# 效果说明:脱离公网API,走本地推理服务

13.4 批量处理与并发优化

python
# 批量并发调用示例
from concurrent.futures import ThreadPoolExecutor
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
questions = ["什么是RAG", "什么是Agent", "什么是Embedding"]

def run(q):
    return llm.invoke(q).content

with ThreadPoolExecutor(max_workers=4) as ex:
    results = list(ex.map(run, questions))

print(results)
# 注意事项:并发过高可能触发限流,需配合重试和速率控制

13.5 Docker 部署模板

dockerfile
# Dockerfile 示例
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

13.6 常见问题(错误代码 + 修正代码)

问题1:API 密钥未配置

python
# 错误代码
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
print(llm.invoke("hello").content)
# 原因说明:未设置 OPENAI_API_KEY,调用会报认证错误
python
# 修正代码
import os
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = "你的API_KEY"
llm = ChatOpenAI(model="gpt-4o-mini")
print(llm.invoke("hello").content)

问题2:向量检索不准确

python
# 常见原因:chunk_size 太大,语义混杂;k 太小召回不足
# 修正方向:减小 chunk_size、增加 k、增加 rerank

问题3:Agent 逻辑混乱

python
# 常见原因:工具描述不清,Prompt 未约束动作顺序
# 修正方向:明确工具输入输出、限制最大迭代、增加中间日志

13.7 高级避坑技巧

  • 降低幻觉:检索增强 + 输出引用来源 + 低温度
  • 降低成本:缓存常见问答、优先小模型、批处理
  • 提升速度:并发 + 流式输出 + 预热连接池
  • 兼容多模型:抽象模型层,统一请求参数

十四、面试高频问题

14.1 LangChain 核心组件有哪些?

  • Prompt、Model、OutputParser、Retriever、Tool、Agent、Memory、Runnable

14.2 LLMChain 和 Agent 的差异?

  • LLMChain 是固定流程,Agent 是动态决策流程。

14.3 RAG 关键优化点?

  • 文档切分、召回质量、重排、上下文压缩、答案引用。

14.4 什么时候用 LangGraph?

  • 多状态、多分支、可恢复流程、人工介入场景。

14.5 3个完整实战案例汇总(从需求到部署)

案例A:智能客服机器人

  • 需求:支持多轮对话与FAQ
  • 实现:ChatModel + 历史记忆 + 工具查询订单
  • 优化:加入敏感词规则、重试和日志
  • 部署:FastAPI + Docker

案例B:企业知识库问答

  • 需求:PDF/TXT 文档检索问答
  • 实现:Loader + Splitter + Chroma + Retriever + QA Chain
  • 优化:chunk 参数调优 + rerank + 来源引用
  • 部署:定时增量索引 + 监控检索命中率

案例C:文档分析工具

  • 需求:批量总结文档并导出报告
  • 实现:批量读取 + LCEL batch + 结构化输出
  • 优化:并发池 + 缓存 + 失败重试
  • 部署:任务队列 + API 查询任务状态

案例D:本地化离线助手

  • 需求:内网可用、数据不出域
  • 实现:本地 LLM(OpenAI 兼容网关)+ 本地向量库
  • 优化:模型量化、索引预构建
  • 部署:单机 Docker Compose

案例E:多工具协同 Agent

  • 需求:自动查询数据并生成日报
  • 实现:ReAct Agent + SQL Tool + Report Tool
  • 优化:权限控制 + 审计日志 + 输出模板
  • 部署:定时任务 + Webhook 推送

核心参数手册(速查)

模型参数

  • model:模型名称,决定能力与成本
  • temperature:随机性,低值更稳定
  • max_tokens:输出长度上限
  • timeout:请求超时
  • top_p:核采样范围(部分模型支持)

检索参数

  • k:召回条数
  • chunk_size:切块大小
  • chunk_overlap:切块重叠
  • search_type:检索策略(similarity/mmr)

Agent 参数

  • max_iterations:最大推理步数
  • verbose:打印中间步骤
  • handle_parsing_errors:解析失败时容错

优质模板(可直接复制改造)

1) 通用问答模板

python
TEMPLATE = """你是专业助手,请基于事实回答:
问题:{question}
要求:
1. 结论先行
2. 分点说明
3. 无依据时明确说明不知道
"""

2) RAG 问答模板

python
RAG_TEMPLATE = """你是知识库助手,只能依据上下文作答。
上下文:
{context}

问题:{question}
请输出:
1) 简短答案
2) 依据片段
"""

3) Agent 工具调用模板

python
AGENT_TEMPLATE = """你可以使用工具解决问题。
策略:
1) 先判断是否需要工具
2) 需要时仅调用最合适工具
3) 输出最终结论与依据
问题:{input}
"""

学习资源推荐


附录:依赖清单示例

txt
langchain
langchain-core
langchain-openai
langchain-community
chromadb
faiss-cpu
pypdf
fastapi
uvicorn
python-dotenv
pydantic

附录:运行步骤(统一)

  1. 创建虚拟环境并安装依赖
  2. 配置 OPENAI_API_KEY(或本地兼容网关)
  3. 运行单文件示例验证模型可用
  4. 再集成 RAG / Tool / Agent
  5. 最后封装 FastAPI 并容器化上线