1. LangChain基础教程:从零构建AI应用开发框架
作为一名长期从事AI应用开发的工程师,我深刻理解初学者在接触LangChain时的困惑。LangChain不仅仅是一个工具库,更是一套完整的LLM应用开发范式。本文将带你系统掌握LangChain的核心概念和实战技巧,这些经验都来自我实际项目中的反复验证。
2. LangChain架构解析
2.1 核心组件构成
LangChain的架构设计遵循"分层解耦"原则,主要包含四个关键层级:
- 模型层:对接不同LLM提供商(如ChatOpenAI/ChatOllama),处理基础推理能力
- 编排层:通过LCEL(LangChain Expression Language)实现流程组装
- 数据层:处理文档加载、分割、向量化存储(如Chroma)
- 能力层:扩展工具调用、记忆、Agent等高级功能
这种分层设计使得开发者可以灵活替换各层实现,比如:
- 模型层可切换不同LLM提供商
- 数据层支持多种向量数据库
- 能力层按需组合功能模块
2.2 典型工作流程
一个完整的LangChain应用通常遵循以下处理流程:
code复制用户输入 → 提示词模板 → LLM推理 → 输出解析 → 结果返回
↑ ↑ ↑
Memory Retriever Callback
这个流程中,各组件通过LCEL的管道符(|)连接,形成可维护的链路。我特别建议新手先掌握这个基础流程,再逐步叠加其他功能。
3. 环境准备与工具配置
3.1 基础环境要求
- Python 3.10+(必须,低版本会有兼容性问题)
- 虚拟环境管理工具(推荐使用uv或conda)
- API Key(如MiniMax等国内可用服务)
- 本地Ollama服务(用于embedding)
提示:建议使用pyenv管理Python版本,避免系统环境污染
3.2 依赖安装
bash复制# 基础依赖
pip install langchain-core langchain-community langchain-openai
# 可选组件
pip install chromadb langchain-chroma # 向量数据库
pip install ollama langchain-ollama # 本地embedding
pip install pypdf # PDF处理
3.3 配置API密钥
创建.env文件管理敏感信息:
ini复制MINIMAX_API_KEY=your_minimax_key
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_EMBED_MODEL=qwen3-embedding:0.6b
使用python-dotenv加载配置:
python复制from dotenv import load_dotenv
load_dotenv()
4. LCEL基础语法精讲
4.1 最小工作流示例
python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
# 定义提示词模板
prompt = ChatPromptTemplate.from_template(
"你是一位{profession}专家。请用3句话介绍{topic}"
)
# 配置LLM
llm = ChatOpenAI(
model="MiniMax-M2.7",
base_url="https://api.minimaxi.com/v1",
temperature=0.7
)
# 构建处理链
chain = prompt | llm | StrOutputParser()
# 调用链
result = chain.invoke({"profession": "技术写作", "topic": "Python装饰器"})
print(result)
4.2 核心组件详解
-
ChatPromptTemplate:
- 支持变量插值(如{profession})
- 可预填部分变量(partial)
- 支持多角色对话模板
-
LLM组件:
- 关键参数:temperature(创造性)、max_tokens(长度)
- 超时控制:request_timeout
- 重试机制:max_retries
-
OutputParser:
- 将LLM输出结构化
- 内置多种解析器(JSON、列表等)
- 支持自定义解析逻辑
4.3 管道操作符原理
|操作符实际上是Runnable接口的串联语法糖,等效于:
python复制chain = prompt.pipe(llm).pipe(StrOutputParser())
管道中的每个组件都必须实现Runnable接口,确保输入输出类型匹配。这是LangChain类型安全的核心机制。
5. 提示词工程实践
5.1 基础模板类型
- 纯文本模板:
python复制from langchain_core.prompts import PromptTemplate
template = PromptTemplate.from_template("总结以下内容:{text}")
- 对话模板:
python复制from langchain_core.prompts import ChatPromptTemplate
chat_template = ChatPromptTemplate.from_messages([
("system", "你是一位专业的技术作家"),
("human", "请解释{concept}")
])
5.2 高级模板技巧
- 模板嵌套:
python复制base_template = """系统角色:{role}
用户问题:{question}"""
prompt = ChatPromptTemplate.from_messages([
("system", base_template),
("human", "{input}")
]).partial(role="技术专家")
- 动态变量注入:
python复制from datetime import datetime
prompt = ChatPromptTemplate.from_messages([
("system", "当前时间:{time}"),
("human", "{query}")
]).partial(time=lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
5.3 实战经验
-
模板管理建议:
- 将常用模板存储在单独文件中
- 使用版本控制管理模板变更
- 对模板进行A/B测试
-
性能优化技巧:
- 对固定内容使用partial预填充
- 批量处理时复用模板实例
- 避免在模板中使用复杂逻辑
6. 记忆机制实现
6.1 基础记忆类型
- 窗口记忆:
python复制from collections import deque
class WindowMemory:
def __init__(self, k=5):
self.memory = deque(maxlen=k)
def add(self, message):
self.memory.append(message)
- 摘要记忆:
python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
summary_prompt = ChatPromptTemplate.from_template("总结对话:{history}")
def summarize(chain, history):
return chain.invoke({"history": "\n".join(history)})
6.2 记忆集成示例
python复制from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import MessagesPlaceholder
# 记忆存储
memory = []
# 提示词集成
prompt = ChatPromptTemplate.from_messages([
("system", "你是有记忆的助手"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
# 使用记忆
memory.append(HumanMessage(content="你好"))
memory.append(AIMessage(content="你好!有什么可以帮您?"))
chain = prompt | llm | StrOutputParser()
result = chain.invoke({
"input": "我刚才问了什么?",
"history": memory
})
6.3 生产环境建议
-
持久化存储:
- 将会话记忆保存到数据库
- 使用Redis等高速缓存
- 定期清理过期记忆
-
记忆优化:
- 对长对话进行摘要压缩
- 设置记忆过期时间
- 基于重要性分级存储
7. 输出解析策略
7.1 内置解析器
- JSON解析:
python复制from langchain_core.output_parsers import JsonOutputParser
parser = JsonOutputParser()
result = parser.invoke('{"name": "John", "age": 30}')
- 列表解析:
python复制from langchain_core.output_parsers import CommaSeparatedListOutputParser
parser = CommaSeparatedListOutputParser()
result = parser.invoke("apple, banana, orange")
7.2 Pydantic集成
python复制from pydantic import BaseModel
from langchain_core.output_parsers import PydanticOutputParser
class Person(BaseModel):
name: str
age: int
parser = PydanticOutputParser(pydantic_object=Person)
result = parser.invoke('{"name": "Alice", "age": 25}')
7.3 自定义解析
python复制from langchain_core.output_parsers import BaseOutputParser
class CustomParser(BaseOutputParser):
def parse(self, text):
# 自定义解析逻辑
return text.upper()
parser = CustomParser()
result = parser.invoke("hello")
8. RAG核心实现
8.1 文档处理流程
- 文档加载:
python复制from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("document.pdf")
pages = loader.load()
- 文本分割:
python复制from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = splitter.split_documents(pages)
- 向量存储:
python复制from langchain_chroma import Chroma
from langchain_ollama import OllamaEmbeddings
embeddings = OllamaEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
8.2 检索增强生成
python复制from langchain_core.runnables import RunnablePassthrough
retriever = vectorstore.as_retriever()
def format_docs(docs):
return "\n\n".join(d.page_content for d in docs)
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
result = rag_chain.invoke("LangChain是什么?")
9. 工具与Agent开发
9.1 工具定义
python复制from langchain_core.tools import tool
@tool
def search(query: str) -> str:
"""搜索引擎工具"""
return f"搜索结果:{query}"
tools = [search]
9.2 Agent创建
python复制from langchain.agents import create_react_agent
agent = create_react_agent(llm, tools, prompt)
9.3 多工具协作
python复制from langchain.agents import Tool
calculator = Tool(
name="Calculator",
func=lambda x: eval(x),
description="数学计算工具"
)
weather = Tool(
name="Weather",
func=lambda loc: f"{loc}的天气:晴",
description="天气查询工具"
)
agent = create_react_agent(llm, [calculator, weather], prompt)
10. 生产环境最佳实践
10.1 性能优化
- 批量处理:
python复制results = chain.batch([
{"input": "query1"},
{"input": "query2"}
])
- 异步处理:
python复制async def process():
return await chain.ainvoke({"input": "query"})
10.2 可观测性
- 回调系统:
python复制from langchain_core.callbacks import StdOutCallbackHandler
chain.invoke(
{"input": "query"},
{"callbacks": [StdOutCallbackHandler()]}
)
- 日志记录:
python复制import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LoggingCallback:
def on_chain_start(self, serialized, inputs, **kwargs):
logger.info(f"Chain started: {inputs}")
10.3 错误处理
- 重试机制:
python复制from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def unreliable_operation():
# 可能失败的操作
pass
- 降级策略:
python复制try:
result = chain.invoke(input)
except Exception:
result = "抱歉,服务暂时不可用"
11. 常见问题排查
-
变量不匹配错误:
- 检查提示词模板中的变量名
- 确认输入字典的键匹配
-
解析失败问题:
- 先检查原始LLM输出
- 添加输出清洗步骤
-
记忆丢失问题:
- 确认记忆存储是否正确持久化
- 检查记忆注入位置
-
检索效果不佳:
- 调整chunk大小
- 尝试不同embedding模型
- 添加元数据过滤
12. 学习路径建议
-
入门阶段:
- 掌握LCEL基础语法
- 理解提示词模板
- 熟悉基本组件连接
-
进阶阶段:
- 实现复杂记忆系统
- 构建RAG应用
- 开发自定义工具
-
专家阶段:
- 优化性能与可靠性
- 实现复杂Agent逻辑
- 设计可扩展架构
在实际项目中,我建议采用迭代开发方式:先构建最小可行产品,再逐步添加高级功能。记住,LangChain的强大之处在于其模块化设计,不要试图一次性实现所有功能。