1. LangChain 快速入门:用通义千问构建你的第一个 AI 应用
最近在开发 AI 应用时,我发现 LangChain 这个框架确实能大幅提升开发效率。作为一个专门用于构建大语言模型应用的框架,它通过组件化和链式调用的方式,让开发者可以像搭积木一样快速构建 AI 应用。今天我就以通义千问为例,带大家快速上手 LangChain。
LangChain 的核心价值在于它提供了标准化的接口和丰富的组件库。无论你使用哪个大模型(ChatGPT、通义千问、Hugging Face 等),都可以用统一的方式调用。这对于需要切换不同模型进行对比测试的场景特别有用。
2. 环境准备与安装
2.1 基础环境配置
首先确保你的 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-turbo: 轻量版,响应快
# qwen-plus: 标准版,平衡性能与效果
# qwen-max: 最强版,效果最好但成本高
model = ChatTongyi(model="qwen-plus")
3.2 构建提示模板
LangChain 提供了强大的提示模板功能。我们可以定义多角色对话模板:
python复制from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个精通{topic}的资深技术专家。"),
("user", "请用三句话解释一下什么是{concept}。")
])
这个模板中有两个变量:
{topic}: 指定专家的专业领域{concept}: 需要解释的概念
提示:LangChain 的提示模板支持更复杂的结构,包括多轮对话、示例对话等。初学者可以先从这种简单结构入手。
3.3 创建处理链
LangChain 的核心概念是 Chain(链),它把多个处理步骤连接起来:
python复制from langchain_core.output_parsers import StrOutputParser
output_parser = StrOutputParser()
chain = prompt | model | output_parser
这个链的工作流程是:
- 将输入数据填充到提示模板
- 将完整提示发送给大模型
- 用输出解析器处理模型返回的结果
StrOutputParser 是最简单的解析器,直接把模型输出转为字符串。LangChain 还提供了其他解析器,比如用于解析 JSON、提取特定字段等。
4. 调用与输出方式
4.1 同步调用
最简单的调用方式是 invoke 方法,它会等待完整响应:
python复制response = chain.invoke({
"topic": "Python",
"concept": "列表"
})
print(response)
输出示例:
code复制列表是Python中的基础数据结构,用于存储有序的元素集合。它使用方括号[]定义,元素可以是不同类型。列表支持动态增删改查,是编程中最常用的容器之一。
4.2 流式输出
对于长文本响应,可以使用流式输出,实现打字机效果:
python复制for chunk in chain.stream({
"topic": "人工智能",
"concept": "神经网络"
}):
print(chunk, end="", flush=True)
这种方式会逐块返回结果,适合需要实时显示的场景,比如聊天应用。
4.3 批量处理
如果需要处理多个输入,可以使用 batch 方法:
python复制responses = chain.batch([
{"topic": "Python", "concept": "字典"},
{"topic": "数学", "concept": "微积分"},
{"topic": "物理", "concept": "量子力学"}
])
for resp in responses:
print(resp)
print("="*30)
5. 高级配置与优化
5.1 模型参数调优
创建模型时可以设置各种参数来调整生成效果:
python复制model = ChatTongyi(
model="qwen-plus",
temperature=0.7, # 控制随机性 (0-1)
top_p=0.9, # 核采样参数
max_tokens=500, # 最大输出长度
n=1 # 生成多个候选结果
)
temperature:值越高输出越随机,适合创意任务;值越低输出越确定,适合事实性问题top_p:只考虑概率累积达到 top_p 的 token,与 temperature 配合使用- 建议开始时使用默认值,根据效果逐步调整
5.2 提示工程技巧
好的提示词能显著提升模型表现。一些实用技巧:
-
明确角色:用 system 消息定义模型角色
python复制("system", "你是一位经验丰富的Python导师,擅长用简单易懂的例子解释概念") -
结构化输出:要求特定格式
python复制("user", "用三句话解释{concept},每句不超过20字") -
提供示例:
python复制("ai", "列表:1. 有序的元素集合 2. 用方括号表示 3. 可以修改") ("user", "现在请解释{concept}")
5.3 错误处理与重试
网络请求可能会失败,建议添加重试逻辑:
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):
return chain.invoke(input_data)
try:
response = safe_invoke(chain, {"topic": "Python", "concept": "生成器"})
except Exception as e:
print(f"请求失败: {e}")
6. 实际应用案例
6.1 技术文档助手
我们可以扩展之前的链,创建一个技术文档助手:
python复制doc_chain = ChatPromptTemplate.from_messages([
("system", "你是一位{topic}专家,擅长编写清晰的技术文档。"),
("user", """为{concept}编写文档,包含:
1. 简短定义(1句话)
2. 核心特点(3-5个要点)
3. 简单代码示例
4. 常见使用场景""")
]) | model | StrOutputParser()
response = doc_chain.invoke({
"topic": "Python编程",
"concept": "装饰器"
})
6.2 多轮对话系统
通过保存对话历史,可以实现多轮对话:
python复制from langchain_core.messages import AIMessage, HumanMessage
history = []
def chat(topic, question):
messages = [
("system", f"你是一位{topic}专家,用友好、专业的语气回答问题。"),
*history,
("user", question)
]
prompt = ChatPromptTemplate.from_messages(messages)
chain = prompt | model | StrOutputParser()
response = chain.invoke({})
history.extend([
HumanMessage(content=question),
AIMessage(content=response)
])
return response
7. 常见问题与解决
7.1 响应速度慢
可能原因及解决方案:
- 模型版本:
qwen-max比qwen-plus慢,如非必要不要用 max 版 - 网络延迟:检查 API 端点,国内用户建议用阿里云国内区域
- 输出长度:限制
max_tokens避免生成过长内容
7.2 输出不符合预期
调试步骤:
- 先打印完整的提示词,确认变量替换正确
python复制print(prompt.format(topic="Python", concept="列表")) - 调整提示词,增加更多约束和示例
- 尝试不同的
temperature和top_p值
7.3 处理敏感内容
如果应用涉及用户输入,建议添加内容过滤:
python复制from langchain.chains import LLMChain
from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.prompts import PromptTemplate
safety_prompt = PromptTemplate(
template="检查以下内容是否包含敏感信息:{text}。只回答是或否",
input_variables=["text"]
)
safety_chain = LLMChain(
llm=model,
prompt=safety_prompt,
output_parser=CommaSeparatedListOutputParser()
)
if safety_chain.run(user_input)[0] == "是":
print("内容包含敏感信息,拒绝处理")
8. 性能优化建议
-
缓存机制:对常见问题缓存响应
python复制from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache()) -
异步调用:提高并发性能
python复制async def async_invoke(): return await chain.ainvoke({"topic": "AI", "concept": "机器学习"}) -
批处理:合并多个请求
python复制responses = await chain.abatch([ {"topic": "Python", "concept": "lambda"}, {"topic": "Math", "concept": "Matrix"} ]) -
监控与日志:记录请求指标
python复制from datetime import datetime def logged_invoke(chain, input_data): start = datetime.now() result = chain.invoke(input_data) duration = (datetime.now() - start).total_seconds() print(f"请求耗时: {duration:.2f}s") return result
通过这些优化,我们的应用可以更高效地处理大量请求,同时保持良好的响应速度。