LangChain入门：用通义千问快速构建AI问答系统

1. LangChain 快速入门：用通义千问构建你的第一个 AI 应用

最近在开发 AI 应用时，我发现 LangChain 这个框架确实能大幅提升开发效率。作为一个专门为大语言模型设计的开发框架，它通过标准化的组件和链式调用，让开发者可以快速搭建 AI 应用原型。今天我就以通义千问为例，带大家从零开始构建一个简单的问答系统。

LangChain 的核心价值在于它提供了一套标准化的开发范式。想象一下，如果没有 LangChain，每次调用不同厂商的模型都需要学习不同的 API 规范，处理不同的返回格式，这会浪费大量时间。而 LangChain 就像是一个万能适配器，把各种大模型的差异都封装起来，让我们可以用统一的方式工作。

2. 环境准备与安装

2.1 基础环境配置

在开始之前，我们需要准备好 Python 环境。建议使用 Python 3.8 或更高版本，并创建一个干净的虚拟环境：

bash复制python -m venv langchain-env
source langchain-env/bin/activate  # Linux/Mac
# 或者
.\langchain-env\Scripts\activate  # Windows

2.2 安装核心依赖

LangChain 采用模块化设计，核心包只包含基础功能，针对不同模型需要安装对应的适配器：

bash复制pip install langchain langchain-community

对于通义千问，我们还需要安装专门的适配器：

bash复制pip install langchain-qwq

提示：如果你要使用其他模型，比如 OpenAI 的 ChatGPT，可以安装 langchain-openai；如果要用 Hugging Face 的模型，则安装 langchain-huggingface。

3. 第一个 LangChain 应用

3.1 初始化模型

首先我们需要获取通义千问的 API Key。登录阿里云控制台，在 DashScope 服务中创建 API Key，然后通过环境变量设置：

python复制import os
os.environ["DASHSCOPE_API_KEY"] = "你的API_KEY"

初始化模型时，可以指定不同的模型版本。通义千问目前提供了多个版本：

python复制from langchain_community.chat_models.tongyi import ChatTongyi

# 使用 qwen-plus 模型
model = ChatTongyi(model="qwen-plus")

# 也可以使用 qwen-turbo 轻量版
# model = ChatTongyi(model="qwen-turbo")

3.2 构建提示模板

LangChain 的 ChatPromptTemplate 让我们可以定义结构化的对话模板：

python复制from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个精通{topic}的资深技术专家。"),
    ("user", "请用三句话解释一下什么是{concept}。")
])

这个模板定义了两个角色：

• system：设定 AI 的角色和专业背景
• user：用户提出的问题

{topic} 和 {concept} 是占位符，在实际调用时会被替换成具体值。

3.3 创建处理链

LangChain 最强大的特性就是可以用管道操作符 | 将各个组件连接成链：

python复制from langchain_core.output_parsers import StrOutputParser

output_parser = StrOutputParser()
chain = prompt | model | output_parser

这个链的工作流程是：

1. 将输入数据填充到提示模板
2. 将完整的提示发送给大模型
3. 用输出解析器处理模型返回的结果

3.4 调用模型

现在我们可以用两种方式调用模型：

完整响应模式（一次性获取全部结果）：

python复制response = chain.invoke({"topic": "Python", "concept": "列表"})
print(response)

流式响应模式（逐字显示，类似打字机效果）：

python复制for chunk in chain.stream({"topic": "人工智能", "concept": "神经网络"}):
    print(chunk, end="", flush=True)

4. 核心组件深入解析

4.1 消息角色详解

在 LangChain 的对话模板中，有三种核心消息类型：

例如，要构建一个带记忆的对话系统，可以这样设计提示模板：

python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content="你是一个乐于助人的AI助手。"),
    AIMessage(content="你好！我是AI助手，有什么可以帮你的？"),  # 上轮AI回复
    HumanMessage(content="{user_input}")  # 当前用户输入
])

4.2 输出解析器

LangChain 提供了多种输出解析器来处理不同的返回格式：

• StrOutputParser：最简单的文本解析器
• JSONOutputParser：解析 JSON 格式响应
• XMLOutputParser：解析 XML 格式响应

例如，如果我们希望模型返回 JSON 格式的数据：

python复制from langchain_core.output_parsers import JSONOutputParser

parser = JSONOutputParser()
chain = prompt | model | parser

response = chain.invoke({"topic": "编程", "concept": "面向对象"})
print(response)  # 直接得到Python字典

4.3 模型参数调优

在初始化模型时，我们可以调整各种参数来优化响应质量：

python复制model = ChatTongyi(
    model="qwen-plus",
    temperature=0.7,  # 控制创造性（0-1）
    top_p=0.9,       # 核采样参数
    max_tokens=500    # 最大输出长度
)

• temperature：值越高输出越随机有创意，值越低输出越确定保守
• top_p：控制采样范围，与 temperature 配合使用
• max_tokens：限制响应长度，防止生成过多内容

5. 实战技巧与常见问题

5.1 性能优化建议

1. 批量处理：如果有多个问题要问，尽量使用 batch 方法一次性发送：

python复制responses = chain.batch([
    {"topic": "Python", "concept": "装饰器"},
    {"topic": "机器学习", "concept": "随机森林"}
])

2. 缓存机制：对于重复的问题，可以添加缓存减少API调用：

python复制from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

set_llm_cache(InMemoryCache())

5.2 错误处理

大模型API调用可能会遇到各种错误，良好的错误处理很重要：

python复制from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_invoke(chain, input_data):
    try:
        return chain.invoke(input_data)
    except Exception as e:
        print(f"调用失败: {str(e)}")
        raise

这个装饰器实现了：

• 最多重试3次
• 指数退避等待（第一次4秒，第二次8秒...）
• 最大等待时间不超过10秒

5.3 调试技巧

当链没有按预期工作时，可以单独检查每个环节的输出：

python复制# 检查填充后的提示
filled_prompt = prompt.invoke({"topic": "Python", "concept": "生成器"})
print(filled_prompt)

# 检查模型原始响应
raw_response = model.invoke(filled_prompt)
print(raw_response)

6. 扩展应用场景

6.1 构建问答知识库

结合向量数据库，可以构建更智能的问答系统：

python复制from langchain_community.vectorstores import FAISS
from langchain_core.embeddings import Embeddings

# 假设我们已经有一些文档
documents = ["文档1内容", "文档2内容"...]

# 创建向量存储
vectorstore = FAISS.from_texts(documents, embedding=Embeddings())

# 构建检索链
retriever = vectorstore.as_retriever()
qa_chain = (
    {"context": retriever, "question": lambda x: x["question"]}
    | prompt
    | model
    | output_parser
)

6.2 多模态应用

LangChain 也支持处理图片等多模态输入：

python复制from langchain_core.messages import HumanMessage
from langchain_community.chat_models.tongyi import ChatTongyi

model = ChatTongyi(model="qwen-vl-plus")  # 视觉语言模型

message = HumanMessage(content=[
    {"type": "text", "text": "描述这张图片"},
    {"type": "image_url", "image_url": "https://example.com/image.jpg"}
])

response = model.invoke([message])

7. 生产环境部署建议

当准备将 LangChain 应用部署到生产环境时，需要考虑以下几点：

1. API 密钥管理：不要将密钥硬编码在代码中，使用环境变量或密钥管理服务：

python复制import os
from dotenv import load_dotenv

load_dotenv()  # 从 .env 文件加载环境变量
api_key = os.getenv("DASHSCOPE_API_KEY")

2. 限流控制：大多数API都有调用频率限制，需要合理控制请求速率：

python复制from tenacity import Retrying, stop_after_delay

# 限制每分钟不超过30次调用
retryer = Retrying(stop=stop_after_delay(60), reraise=True)
for _ in range(30):
    retryer(lambda: chain.invoke(input_data))

3. 监控与日志：记录每次调用的输入输出，便于问题排查：

python复制import logging
from datetime import datetime

logging.basicConfig(filename='llm.log', level=logging.INFO)

def log_invocation(input_data, output):
    logging.info(f"{datetime.now()} - Input: {input_data}")
    logging.info(f"{datetime.now()} - Output: {output[:200]}...")  # 截断长输出

response = chain.invoke(input_data)
log_invocation(input_data, response)

通过这些实践，我成功将多个 LangChain 应用部署到了生产环境。最大的体会是：LangChain 确实能显著提升开发效率，但要构建健壮的生产级应用，还需要在错误处理、监控和性能优化上下功夫。