LangChain 框架详解
1. 什么是 LangChain
LangChain 是一个用于构建基于大语言模型(LLM)应用程序的框架。它提供了一套工具和抽象,使开发者能够轻松地将语言模型与其他数据源和计算工具结合起来,构建复杂的AI应用。
核心特性
- 模块化设计:提供可组合的组件
- 链式调用:支持复杂的工作流程
- 多模型支持:兼容多种LLM提供商
- 内存管理:支持对话历史和上下文管理
- 工具集成:可以集成外部API和工具
2. 安装和环境配置
基础安装
bash
pip install langchain安装特定集成
bash
# OpenAI 集成
pip install langchain-openai
# 向量数据库集成
pip install langchain-chroma
pip install langchain-pinecone
# 其他常用工具
pip install langchain-community环境变量配置
bash
# 设置 OpenAI API Key
export OPENAI_API_KEY="your-api-key-here"
# 其他API密钥
export ANTHROPIC_API_KEY="your-anthropic-key"
export GOOGLE_API_KEY="your-google-key"3. 核心概念
3.1 LLMs 和 Chat Models
python
from langchain_openai import OpenAI, ChatOpenAI
# 基础LLM
llm = OpenAI(temperature=0.7)
result = llm.invoke("什么是人工智能?")
# 聊天模型
chat_model = ChatOpenAI(model="gpt-3.5-turbo")
messages = [
("system", "你是一个有用的AI助手"),
("human", "解释一下机器学习")
]
result = chat_model.invoke(messages)3.2 Prompts 和 Prompt Templates
python
from langchain.prompts import PromptTemplate, ChatPromptTemplate
# 简单提示模板
prompt = PromptTemplate(
input_variables=["topic"],
template="请写一篇关于{topic}的简短介绍"
)
# 聊天提示模板
chat_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的{domain}专家"),
("human", "请解释{concept}的概念")
])3.3 Chains(链)
python
from langchain.chains import LLMChain
from langchain_openai import OpenAI
# 创建简单链
llm = OpenAI()
prompt = PromptTemplate(
input_variables=["question"],
template="问题:{question}\n答案:"
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run("什么是深度学习?")4. 实际应用示例
4.1 文档问答系统
python
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_openai import OpenAI
# 1. 加载文档
loader = TextLoader("document.txt")
documents = loader.load()
# 2. 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)
# 3. 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)
# 4. 创建问答链
qa = RetrievalQA.from_chain_type(
llm=OpenAI(),
chain_type="stuff",
retriever=vectorstore.as_retriever()
)
# 5. 提问
answer = qa.run("文档中提到了什么重要概念?")4.2 对话记忆系统
python
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import OpenAI
# 创建记忆组件
memory = ConversationBufferMemory()
# 创建对话链
conversation = ConversationChain(
llm=OpenAI(),
memory=memory,
verbose=True
)
# 进行对话
response1 = conversation.predict(input="我叫张三")
response2 = conversation.predict(input="我的名字是什么?")4.3 工具使用(Agent)
python
from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain_openai import OpenAI
import requests
def search_web(query):
"""简单的网络搜索工具"""
# 这里可以集成真实的搜索API
return f"搜索结果:{query}"
def calculate(expression):
"""计算工具"""
try:
return str(eval(expression))
except:
return "计算错误"
# 定义工具
tools = [
Tool(
name="搜索",
func=search_web,
description="用于搜索网络信息"
),
Tool(
name="计算器",
func=calculate,
description="用于数学计算"
)
]
# 创建代理
agent = initialize_agent(
tools,
OpenAI(),
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
# 使用代理
result = agent.run("计算 15 * 23 的结果")5. 高级功能
5.1 自定义链
python
from langchain.chains.base import Chain
from typing import Dict, List
class CustomChain(Chain):
"""自定义链示例"""
@property
def input_keys(self) -> List[str]:
return ["input_text"]
@property
def output_keys(self) -> List[str]:
return ["output_text"]
def _call(self, inputs: Dict[str, str]) -> Dict[str, str]:
# 自定义处理逻辑
processed_text = inputs["input_text"].upper()
return {"output_text": processed_text}
# 使用自定义链
custom_chain = CustomChain()
result = custom_chain({"input_text": "hello world"})5.2 流式输出
python
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_openai import OpenAI
# 创建带流式输出的LLM
llm = OpenAI(
streaming=True,
callbacks=[StreamingStdOutCallbackHandler()]
)
# 流式生成
response = llm.invoke("写一首关于春天的诗")6. 最佳实践
6.1 错误处理
python
from langchain.schema import OutputParserException
try:
result = chain.run(input_text)
except OutputParserException as e:
print(f"解析错误: {e}")
except Exception as e:
print(f"其他错误: {e}")6.2 性能优化
python
# 使用缓存
from langchain.cache import InMemoryCache
import langchain
langchain.llm_cache = InMemoryCache()
# 批量处理
inputs = [{"question": f"问题{i}"} for i in range(10)]
results = chain.batch(inputs)6.3 调试和监控
python
# 启用详细输出
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
# 使用回调函数
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
result = chain.run("测试问题")
print(f"总tokens: {cb.total_tokens}")
print(f"成本: ${cb.total_cost}")7. 常见问题和解决方案
问题1:API限制
python
# 添加重试机制
from langchain.llms import OpenAI
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 call_llm_with_retry(llm, prompt):
return llm(prompt)问题2:内存管理
python
# 使用滑动窗口内存
from langchain.memory import ConversationSummaryBufferMemory
memory = ConversationSummaryBufferMemory(
llm=OpenAI(),
max_token_limit=1000
)8. 总结
LangChain 是一个强大的框架,它简化了基于LLM的应用开发。通过其模块化的设计,开发者可以快速构建复杂的AI应用,包括聊天机器人、文档问答系统、智能代理等。
