一、前置准备（依赖库+环境配置）

上次分享了LangChain入门指南后，很多朋友私信我：“入门案例已经跑通了，接下来该学什么？”“怎么用LangChain做更实用的AI应用？”

其实LangChain的入门不难，难的是从“会用”到“活用”——摆脱简单的对话、文本问答，掌握进阶组件和实战技巧，才能真正把它用到工作和项目中。

今天这篇进阶文，就衔接上一篇入门内容，避开冗余理论，直接实战：

先梳理入门到进阶的核心跃迁点，再拆解3个高频进阶案例（含完整可运行代码，适配阿里千问，复制就能跑），帮你吃透LangChain的进阶能力，从初学者快速成长为能独立搭建实战应用的开发者。

（全文干货+代码，建议收藏，搭配上一篇入门文食用，效果翻倍；代码已规避版本兼容问题，新手也能零踩坑）

1.1 依赖库安装（终端执行）

核心作用：安装进阶案例所需的所有依赖库，固定版本以避免兼容问题，包含环境管理、向量库、文档解析、模型依赖等，终端复制对应命令执行即可。

pip install python-dotenv==1.0.0 faiss-cpu==1.7.4 python-docx==1.1.0 tiktoken==0.5.2 langchain==0.1.10 dashscope==1.14.0

1.2 .env文件配置（同级目录新建.env文件，无后缀）

核心作用：工程化管理阿里千问API密钥，避免明文泄露，需在新建的.env文件中，填写个人的AccessKey ID和AccessKey Secret（替换占位符）。

DASHSCOPE_ACCESS_KEY_ID=你的AccessKey ID
DASHSCOPE_ACCESS_KEY_SECRET=你的AccessKey Secret

二、通用函数（所有案例均可复用）

核心作用：封装所有案例共用的基础操作，无需重复编写，对应代码位置包含5个核心部分，分别是：

• 导入所需核心依赖包（环境、模型、文本分割、向量库相关）；

• 读取.env文件中的阿里千问API密钥，初始化密钥变量；

• init_llm函数：初始化阿里千问模型，可调整温度、最大输出token数等参数；

• init_text_splitter函数：初始化文本分割工具，适配中文文本，可调整分割块大小和重叠长度；

• init_vector_store函数：初始化向量库，将文本块转为向量并保存到本地，方便后续复用；

• load_local_vector_store函数：加载本地已保存的向量库，避免重复生成，提升效率。

import os
from dotenv import load_dotenv
from langchain_community.llms import DashScope
from langchain_community.embeddings import DashScopeEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import FAISS

load_dotenv()
access_key_id = os.getenv("DASHSCOPE_ACCESS_KEY_ID")
access_key_secret = os.getenv("DASHSCOPE_ACCESS_KEY_SECRET")

def init_llm(temperature=0.5, max_tokens=1000):
    llm = DashScope(
        model_name="qwen-turbo",
        temperature=temperature,
        max_tokens=max_tokens,
        dashscope_api_key=access_key_secret
    )
    return llm

def init_text_splitter(chunk_size=500, chunk_overlap=50):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", "。", "，", "；"]
    )
    return text_splitter

def init_vector_store(text_chunks, embeddings=None):
    if embeddings is None:
        embeddings = DashScopeEmbeddings(
            model="text-embedding-v2",
            dashscope_api_key=access_key_secret
        )
    vector_store = FAISS.from_documents(text_chunks, embeddings)
    vector_store.save_local("faiss_vector_store")
    return vector_store

def load_local_vector_store(embeddings=None):
    if embeddings is None:
        embeddings = DashScopeEmbeddings(
            model="text-embedding-v2",
            dashscope_api_key=access_key_secret
        )
    vector_store = FAISS.load_local("faiss_vector_store", embeddings, allow_dangerous_deserialization=True)
    return vector_store

三、案例1：进阶版多文档问答

这3个案例，是进阶阶段最高频、最实用的实战场景，循序渐进，从“多文档问答”（基础进阶），到“结构化输出批量生成”（实战常用），再到“AI智能体入门”（高阶能力），每个案例都衔接上一篇入门内容，且包含代码注释、运行说明和避坑要点。

所有代码都适配阿里千问，无需科学上网，替换.env文件中的密钥即可直接运行。

核心作用：实现多文档（PDF、Word、TXT）问答，支持批量加载、向量库复用、来源标注，对应代码位置包含7个核心部分，分别是：

• 导入多文档问答所需依赖（文档加载、检索链、提示词模板、异常捕获相关）；

• 调用通用函数，初始化千问模型（调整温度适配结构化问答）和文本分割工具（调整参数适配多文档）；

• load_multiple_documents函数：批量加载多格式文档（PDF、Word、TXT），标注文档来源，增加异常捕获避免单个文件加载失败；

• 设置多文档路径（替换占位符为个人文档路径），加载文档并分割为文本块，打印分割结果；

• 尝试加载本地已保存的向量库，无本地向量库则初始化并保存，提升复用效率；

• 定义多文档问答提示词模板（明确回答要求、来源标注规则），初始化检索式问答链（调整参数提升检索精度）；

• 搭建循环交互逻辑，支持用户连续提问，输出问答结果及来源详情，输入“退出”可结束交互。

from langchain.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import traceback

llm = init_llm(temperature=0.3)
text_splitter = init_text_splitter(chunk_size=600, chunk_overlap=60)

def load_multiple_documents(file_paths):
    documents = []
    for file_path in file_paths:
        try:
            if file_path.endswith(".pdf"):
                loader = PyPDFLoader(file_path)
            elif file_path.endswith(".docx"):
                loader = Docx2txtLoader(file_path)
            elif file_path.endswith(".txt"):
                loader = TextLoader(file_path, encoding="utf-8")
            else:
                print(f"不支持的文件格式：{file_path}")
                continue
            doc = loader.load()
            for d in doc:
                d.metadata["source"] = file_path
            documents.extend(doc)
            print(f"成功加载文件：{file_path}，共{len(doc)}页")
        except Exception as e:
            print(f"加载文件{file_path}失败：{str(e)}")
            traceback.print_exc()
    return documents

file_paths = [
    "文档1.pdf",
    "文档2.docx",
    "文档3.txt"
]
documents = load_multiple_documents(file_paths)
text_chunks = text_splitter.split_documents(documents)
print(f"多文档分割完成，共生成{len(text_chunks)}个文本块")

try:
    vector_store = load_local_vector_store()
    print("成功加载本地向量库")
except:
    print("本地无向量库，开始初始化向量库...")
    vector_store = init_vector_store(text_chunks)
    print("向量库初始化并保存完成")

prompt_template = """
你是一个专业的多文档问答助手，严格按照以下要求回答：
1. 仅基于提供的多文档内容回答问题，不编造任何信息；
2. 回答时，先给出核心答案，再标注回答来源（格式：来源：XXX文件，页码：XXX）；
3. 若多个文件包含相关内容，全部标注来源；若文档中无相关内容，直接回复“文档中未找到相关信息”；
4. 回答简洁明了，分点清晰（复杂问题可分点，简单问题一句话概括）。

多文档相关内容：
{context}

用户问题：{question}

回答：
"""
prompt = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="map_reduce",
    retriever=vector_store.as_retriever(
        k=3,
        search_kwargs={"score_threshold": 0.7}
    ),
    chain_type_kwargs={"prompt": prompt},
    return_source_documents=True
)

print("\n=== 多文档问答演示（输入'退出'结束）===")
while True:
    user_input = input("你：")
    if user_input.lower() == "退出":
        print("对话结束")
        break
    result = qa_chain({"query": user_input})
    print(f"\nAI：{result['result']}")
    print("\n回答来源详情：")
    for idx, doc in enumerate(result["source_documents"], 1):
        source = doc.metadata.get("source", "未知来源")
        page = doc.metadata.get("page", "未知页码")
        content = doc.page_content[:150] + "..."
        print(f"{idx}. 来源：{source}，页码：{page}，相关内容：{content}")
    print("\n" + "-"*50)

四、案例2：结构化输出进阶（批量生成+格式校验）

核心作用：批量生成复杂结构化数据，实现格式严格校验、异常重试、数据持久化，对应代码位置包含8个核心部分，分别是：

• 导入结构化输出所需依赖（输出解析、提示词模板、链、格式校验、时间、JSON处理相关）；

• 调用通用函数，初始化千问模型（调整温度和最大token数，适配结构化生成）；

• 定义结构化格式模型（Author嵌套模型、Book核心模型），明确字段类型、约束条件和示例，实现严格校验；

• 初始化JSON输出解析器，基于定义的结构化模型，实现格式校验和解析；

• 定义提示词模板，明确结构化生成要求、格式约束，避免格式错误；

• 初始化生成链，串联提示词、模型、输出解析器，实现“提示→生成→校验”一体化；

• batch_generate_books函数：实现批量生成逻辑，支持设置生成数量和重试次数，格式错误时自动重试；

• 调用批量生成函数，生成指定数量的结构化数据，格式化输出生成结果，清晰展示每一条数据的核心信息；

• 将批量生成的结构化数据保存到JSON文件，实现数据持久化，方便后续复用（如填充表格、生成报告）。

from langchain_core.output_parsers import JsonOutputParser, OutputParserException
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import LLMChain
from pydantic import BaseModel, Field
import time
import json

llm = init_llm(temperature=0.3, max_tokens=1500)

class Author(BaseModel):
    name: str = Field(description="书籍作者姓名，必须是中文", examples=["张三"])
    gender: str = Field(description="作者性别，只能是'男'或'女'或'未知'", examples=["男"])
    introduction: str = Field(description="作者简介，50-100字，简洁明了", examples=["张三，资深AI工程师，从事LLM应用开发5年，出版多本AI相关书籍"])

class Book(BaseModel):
    book_name: str = Field(description="书籍名称，必须是中文，不能包含英文", examples=["LangChain实战指南"])
    author: Author = Field(description="作者信息，嵌套结构，严格按照Author模型格式填写")
    publish_year: int = Field(description="出版年份，必须是整数，范围在2000-2026之间", examples=[2025])
    category: str = Field(description="书籍分类，只能是'人工智能'、'历史'、'科幻'、'文学'、'编程'中的一种", examples=["人工智能"])
    introduction: str = Field(description="书籍简介，100-200字，详细说明书籍核心内容", examples=["本书从初学者视角出发，拆解LangChain的核心组件和实战案例，包含完整代码，帮助新手快速上手LLM应用开发，适配阿里千问等国产模型。"])
    price: float = Field(description="书籍价格，必须是浮点数，范围在20.0-100.0之间", examples=[69.9])

parser = JsonOutputParser(pydantic_object=Book)

prompt_template = ChatPromptTemplate.from_messages([
    (
        "system",
        """你是一个专业的书籍信息结构化生成助手，严格按照以下要求执行：
1. 严格按照指定的JSON格式生成书籍信息，字段类型、约束必须符合要求（比如出版年份是整数，价格是浮点数）；
2. 批量生成时，每本书的信息必须独立、不重复，分类均匀分布在指定分类中；
3. 作者简介、书籍简介必须符合字数要求，语言流畅，符合中文表达习惯；
4. 若生成的格式不符合要求，会被强制重试，直到符合要求为止；
5. 只输出JSON格式，不包含任何额外文字、注释，否则会报错。

输出格式要求（严格遵守）：
{format_instructions}
"""
    ),
    (
        "human",
        "请批量生成{num}本中文书籍的信息，要求覆盖不同分类，每本书的信息完整、真实可信，符合上述要求。"
    )
])

chain = prompt_template | llm | parser

def batch_generate_books(num=5, retry=3):
    books_list = []
    print(f"开始批量生成{num}本中文书籍信息...")
    for i in range(num):
        print(f"\n生成第{i+1}本书信息：")
        for attempt in range(retry):
            try:
                result = chain.invoke({
                    "num": 1,
                    "format_instructions": parser.get_format_instructions()
                })
                books_list.append(result)
                print(f"生成成功：{result['book_name']}")
                break
            except OutputParserException as e:
                print(f"第{attempt+1}次生成失败，格式错误：{str(e)}，正在重试...")
                time.sleep(1)
            except Exception as e:
                print(f"第{attempt+1}次生成失败，错误：{str(e)}，正在重试...")
                time.sleep(1)
        else:
            print(f"第{i+1}本书生成失败，已达到最大重试次数")
    return books_list

batch_books = batch_generate_books(num=5, retry=3)

print("\n=== 批量生成书籍信息完成 ===")
for idx, book in enumerate(batch_books, 1):
    print(f"\n{idx}. 书籍名称：{book['book_name']}")
    print(f"   作者：{book['author']['name']}（{book['author']['gender']}）")
    print(f"   出版年份：{book['publish_year']}年")
    print(f"   分类：{book['category']}")
    print(f"   价格：{book['price']}元")
    print(f"   作者简介：{book['author']['introduction']}")
    print(f"   书籍简介：{book['introduction']}")

with open("batch_books.json", "w", encoding="utf-8") as f:
    json.dump(batch_books, f, ensure_ascii=False, indent=4)
print("\n批量生成的书籍信息已保存到 batch_books.json 文件中")

五、案例3：AI智能体入门（自主工具调用）

核心作用：实现AI自主判断需求、调用合适工具完成复杂任务，支持多轮对话记忆，对应代码位置包含8个核心部分，分别是：

• 导入AI智能体所需依赖（智能体、工具、记忆组件、检索链、网络请求相关）；

• 调用通用函数，初始化千问模型（调整温度，平衡决策理性和灵活性）；

• get_weather函数：自定义天气查询工具，调用免费天气API，根据城市名称返回格式化天气信息，增加异常处理；

• 加载本地向量库（需先运行案例1生成），初始化多文档检索链，作为第二个自定义工具；

• 注册工具列表，将天气查询工具和多文档检索工具注册到LangChain，明确每个工具的名称、对应函数和适用场景描述（供AI判断调用）；

• 初始化总结式记忆组件，压缩对话上下文，节省token，支持多轮对话上下文关联；

• 初始化AI智能体，绑定工具和记忆，设置智能体类型（适配多轮对话+工具调用），开启详细日志便于调试；

• 搭建循环交互逻辑，支持用户连续提问，AI自主判断需求并调用对应工具，输入“退出”可结束交互，附带提问示例供参考。

from langchain.agents import AgentType, initialize_agent, Tool
from langchain.memory import ConversationSummaryMemory
from langchain.chains import RetrievalQA
import requests

llm = init_llm(temperature=0.5)

def get_weather(city: str) -> str:
    try:
        url = f"https://v0.yiketianqi.com/api?unescape=1&version=v61&appid=43656176&appsecret=I42og6Lm&city={city}"
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        weather_data = response.json()
        weather_info = f"""
{city}当天天气信息：
1. 天气状况：{weather_data['wea']}
2. 温度范围：{weather_data['tem1']}℃ ~ {weather_data['tem2']}℃
3. 风向风力：{weather_data['win']} {weather_data['win_speed']}
4. 湿度：{weather_data['humidity']}%
5. 空气质量：{weather_data['air_quality']}（AQI：{weather_data['air']}）
6. 穿衣建议：{weather_data['air_tips']}
        """
        return weather_info.strip()
    except Exception as e:
        return f"天气查询失败：{str(e)}，请检查城市名称是否正确，或稍后重试。"

try:
    vector_store = load_local_vector_store()
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="map_reduce",
        retriever=vector_store.as_retriever(k=3, search_kwargs={"score_threshold": 0.7}),
        return_source_documents=False
    )
except:
    print("本地无向量库，请先运行案例1，生成多文档向量库后再运行本案例")
    exit()

tools = [
    Tool(
        name="WeatherQuery",
        func=get_weather,
        description="""用于查询中国城市的当天天气信息，当用户的问题包含“天气”“温度”“穿衣建议”等关键词，且明确提到城市名称时，调用此工具。
        注意：必须传入中文城市名称（如“北京”），不能传入英文或拼音。"""
    ),
    Tool(
        name="DocumentRetrieval",
        func=qa_chain.run,
        description="""用于回答用户关于多文档内容的问题，当用户的问题是询问文档中的信息、知识点，且不涉及天气时，调用此工具。
        注意：只能基于已加载的多文档内容回答，不能编造信息；若文档中无相关内容，直接回复“文档中未找到相关信息”。"""
    )
]

memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True,
    handle_parsing_errors="检查你的回答格式，确保符合要求，不要包含任何额外文字，只输出最终回答。"
)

print("\n=== AI智能体演示（输入'退出'结束）===")
print("提示：可尝试提问：'北京今天天气怎么样？'、'文档中提到的LangChain核心组件有哪些？'、'上海今天适合穿什么？'")
while True:
    user_input = input("你：")
    if user_input.lower() == "退出":
        print("对话结束")
        break
    response = agent.run(input=user_input)
    print(f"\nAI：{response}\n")
    print("-"*60)