Skip to content

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应用,包括聊天机器人、文档问答系统、智能代理等。