1. LangChain 入门:快速构建大语言模型应用
作为一名长期从事AI应用开发的工程师,我最近在多个项目中使用了LangChain框架,它确实大幅提升了我们与大语言模型(LLM)集成的效率。今天我就来分享如何用LangChain快速搭建一个可用的LLM应用,以通义千问(Qwen)为例进行演示。
LangChain的核心价值在于它提供了一套标准化的组件和连接方式,让我们可以像搭积木一样构建LLM应用。不同于直接调用API,LangChain通过"链(Chain)"的概念将提示词构建、模型调用、输出解析等环节串联起来,既保持了灵活性又降低了开发复杂度。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 环境准备与安装
2.1 基础环境配置
首先确保你的Python环境是3.8或更高版本。我推荐使用conda或venv创建独立的Python环境:
bash复制python -m venv langchain-env
source langchain-env/bin/activate # Linux/Mac
# 或者
.\langchain-env\Scripts\activate # Windows
2.2 安装LangChain核心库
LangChain采用模块化设计,核心包只包含基础功能,特定模型的集成需要额外安装:
bash复制pip install langchain langchain-community
根据你要使用的模型,还需要安装对应的集成包。以通义千问为例:
bash复制pip install langchain-qwq
其他常用模型的安装命令:
bash复制# OpenAI
pip install langchain-openai
# Hugging Face
pip install langchain-huggingface
# Anthropic Claude
pip install langchain-anthropic
提示:生产环境中建议使用requirements.txt或poetry管理依赖,避免版本冲突。
3. 构建第一个LangChain应用
3.1 获取API密钥
使用通义千问需要先在阿里云DashScope平台获取API Key。注册后可以在控制台找到:
python复制import os
os.environ["DASHSCOPE_API_KEY"] = "你的API_KEY"
3.2 初始化模型
LangChain为不同模型提供了统一的接口。初始化千问模型很简单:
python复制from langchain_community.chat_models.tongyi import ChatTongyi
model = ChatTongyi(
model="qwen-plus", # 也可用"qwen-turbo"等
temperature=0.7, # 控制创造性,0-1之间
top_p=0.9 # 采样参数
)
关键参数说明:
model: 指定模型版本,不同版本在性能和价格上有差异temperature: 值越高输出越随机有创意,越低越确定保守top_p: 核采样参数,控制生成多样性
3.3 构建提示模板
LangChain的ChatPromptTemplate让提示词管理更规范:
python复制from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个精通{topic}的资深技术专家。"),
("user", "请用三句话解释一下什么是{concept}。")
])
这个模板定义了两个角色:
system: 设定AI的角色和专业背景user: 用户实际提问内容
{topic}和{concept}是占位符,实际调用时会动态替换。
3.4 创建处理链
LangChain的核心抽象就是"链(Chain)",将各组件连接起来:
python复制from langchain_core.output_parsers import StrOutputParser
output_parser = StrOutputParser()
chain = prompt | model | output_parser
这个管道的工作流程是:
- 输入数据填充提示模板
- 完整提示发送给模型
- 原始输出解析为字符串
3.5 调用模型
有两种主要调用方式:
一次性获取完整响应:
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的提示模板支持多种消息角色:
| 角色 | 类名 | 用途说明 |
|---|---|---|
| system | SystemMessage | 设定AI角色和行为准则 |
| user | HumanMessage | 用户实际提问内容 |
| ai | AIMessage | 模型之前的回复(多轮对话用) |
| function | FunctionMessage | 函数调用相关消息 |
多轮对话示例:
python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个专业的技术顾问"),
HumanMessage(content="Python中的装饰器是什么?"),
AIMessage(content="装饰器是用来修改函数行为的特殊语法。"),
HumanMessage(content="能举个实际例子吗?")
])
4.2 模型调用参数调优
不同模型支持的参数略有差异,但常见的有:
python复制model = ChatTongyi(
model="qwen-plus",
temperature=0.7, # 创造性(0-1)
top_p=0.9, # 多样性控制
max_tokens=1024, # 最大输出长度
frequency_penalty=0, # 重复惩罚(-2到2)
presence_penalty=0 # 话题新颖度(-2到2)
)
实践建议:对于技术问答,temperature建议0.3-0.7;创意写作可以0.8以上。
4.3 输出解析器
除了StrOutputParser,LangChain还提供多种解析器:
JSONOutputParser: 解析JSON格式输出XMLOutputParser: 解析XML格式CommaSeparatedListOutputParser: 解析逗号分隔列表- 自定义解析器:继承
BaseOutputParser
JSON解析示例:
python复制from langchain_core.output_parsers import JSONOutputParser
parser = JSONOutputParser()
chain = prompt | model | parser
response = chain.invoke({
"topic": "编程",
"concept": "REST API"
}) # 返回字典而非字符串
5. 实战技巧与问题排查
5.1 性能优化建议
-
批量处理:使用
batch()同时处理多个输入python复制inputs = [ {"topic": "Python", "concept": "生成器"}, {"topic": "AI", "concept": "Transformer"} ] responses = chain.batch(inputs) -
异步调用:对于Web应用使用
ainvokepython复制response = await chain.ainvoke(...) -
缓存:对相同输入缓存结果
python复制from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())
5.2 常见错误排查
问题1:API密钥无效
- 检查环境变量名是否正确
- 确认密钥未过期
- 验证API服务区域
问题2:模型返回意外内容
- 检查temperature是否过高
- 验证提示模板是否清晰
- 添加更具体的系统提示
问题3:输出解析失败
- 确保选择了正确的解析器
- 让模型返回更结构化的响应
- 添加输出格式指令到提示词
5.3 生产环境最佳实践
- 错误处理:
python复制from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def safe_invoke(chain, input):
try:
return chain.invoke(input)
except Exception as e:
log_error(e)
return "服务暂时不可用"
- 监控:记录调用指标
- 延迟
- 令牌使用量
- 成功率
- 限流:避免突发流量
python复制from langchain.adapters.openai import TokenBucket
limiter = TokenBucket(rate=5) # 5次/秒
chain = limiter | prompt | model | parser
6. 扩展应用场景
6.1 文档问答系统
结合文档加载器和向量数据库:
python复制from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
loader = WebBaseLoader("https://example.com/doc")
docs = loader.load()
embeddings = OpenAIEmbeddings()
db = FAISS.from_documents(docs, embeddings)
retriever = db.as_retriever()
qa_chain = create_qa_chain(model, retriever)
6.2 智能客服机器人
实现多轮对话记忆:
python复制from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
chain = (
{"input": lambda x: x["input"],
"history": lambda x: memory.load_memory_variables(x)["history"]}
| prompt
| model
| parser
)
response = chain.invoke({"input": "什么是Python装饰器?"})
memory.save_context({"input": "什么是Python装饰器?"}, {"output": response})
6.3 数据分析助手
让LLM生成和执行代码:
python复制from langchain_experimental.agents import create_pandas_dataframe_agent
import pandas as pd
df = pd.read_csv("data.csv")
agent = create_pandas_dataframe_agent(model, df)
response = agent.invoke("哪个产品的销售额最高?")
在实际项目中,LangChain的这些组件可以灵活组合。我最近在一个客户项目中,就用检索链(RAG)配合自定义提示模板,将客服响应准确率提升了40%。关键是要理解每个组件的职责,然后像搭积木一样构建适合自己业务的解决方案。
