Computer Science

WangMC Blog

LangChain基础使用

发表于 分类于
本文字数： 14k 阅读时长 ≈ 13 分钟

结合官方文档，利用ai整理收集，作为了解

LangChain 使用指南

目录

  1. 概述

  2. 核心概念

  3. 安装与环境配置

  4. 核心模块详解

  5. 实际应用示例

  6. 最佳实践

  7. 常见问题

  8. 参考资源

概述

LangChain 是一个基于语言模型的应用程序开发框架，旨在帮助开发者快速构建强大的 LLM 应用。它提供了一套完整的工具和组件，使开发者能够轻松地将语言模型与外部数据源、API 和其他工具集成。

为什么选择 LangChain？

  • 模块化设计：提供了丰富的组件，可以灵活组合

  • 多模型支持：支持主流的 LLM 提供商（OpenAI、Anthropic、DeepSeek 等）

  • 数据集成：轻松连接外部知识库和数据源

  • 代理能力：支持构建能够使用工具的智能代理

  • 会话管理：内置记忆机制，支持多轮对话

核心概念

1. 模型 I/O (Model I/O)

负责处理语言模型的输入和输出，包括：

  • LLMs：文本输入文本输出的语言模型

  • ChatModels：结构化消息处理的聊天模型

  • Embeddings：将文本转换为向量表示

2. 数据连接 (Data Connection)

处理与外部数据的交互：

  • Document Loaders：从各种来源加载文档

  • Text Splitters：将文档分割成可处理的片段

  • Vector Stores：存储和检索向量嵌入

  • Retrievers：从向量存储中检索相关文档

3. 记忆 (Memory)

管理对话历史和状态：

  • Conversation Memory：存储对话历史

  • Entity Memory：跟踪对话中的实体信息

  • Knowledge Graphs：结构化知识存储

4. 链 (Chains)

将多个组件组合成工作流：

  • Simple Chains：线性执行的简单链

  • Sequential Chains：顺序执行多个链

  • Parallel Chains：并行执行多个链

  • Router Chains：根据条件选择执行路径

5. 代理 (Agents)

能够自主决策和使用工具的智能体：

  • Tool Calling：调用外部工具和 API

  • Reasoning：推理和决策能力

  • Planning：复杂任务的规划和执行

安装与环境配置

基础安装

1
2
3
4
5
# 安装核心包
pip install langchain

# 完整安装（包含所有依赖）
pip install langchain langchain-community langchain-core langchain-openai

环境变量配置

1
2
3
4
5
# 设置 API 密钥
export OPENAI_API_KEY="your-api-key"
export LANGCHAIN_API_KEY="your-langchain-api-key"
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_PROJECT="your-project-name"

支持的模型提供商

1
2
3
4
5
6
7
8
9
10
11
# OpenAI
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# Anthropic
from langchain_anthropic import ChatAnthropic

# DeepSeek
from langchain_openai import ChatOpenAI  # DeepSeek 兼容 OpenAI API

# 国产模型
from langchain_community.llms import AlibabaLLM, ZhipuAI

核心模块详解

1. 模型 I/O

聊天模型示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

# 初始化模型
model = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
    max_tokens=1024
)

# 直接调用模型
response = model.invoke([
    HumanMessage(content="你好，我是小明")
])
print(response.content)

# 多轮对话
response = model.invoke([
    HumanMessage(content="你好，我是小明"),
    AIMessage(content="你好小明！有什么我可以帮助你的吗？"),
    HumanMessage(content="我的名字是什么？")
])
print(response.content)

嵌入模型示例

1
2
3
4
5
6
7
8
9
10
from langchain_openai import OpenAIEmbeddings

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 生成文本嵌入
text = "这是一个测试句子"
embedding = embeddings.embed_query(text)
print(f"嵌入向量维度: {len(embedding)}")
print(f"嵌入向量前5个值: {embedding[:5]}")

2. 数据连接

文档加载和处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 加载 PDF 文档
loader = PyPDFLoader("example.pdf")
documents = loader.load()

# 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)

docs = text_splitter.split_documents(documents)
print(f"分割后的文档片段数量: {len(docs)}")
print(f"第一个片段: {docs[0].page_content[:200]}...")

向量存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from langchain.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 初始化嵌入模型
embeddings = OpenAIEmbeddings()

# 创建向量存储
vectorstore = FAISS.from_documents(docs, embeddings)

# 保存和加载向量存储
vectorstore.save_local("faiss_index")
loaded_vectorstore = FAISS.load_local("faiss_index", embeddings)

# 相似性检索
query = "如何使用 LangChain 构建 RAG 应用？"
docs = vectorstore.similarity_search(query, k=3)

for i, doc in enumerate(docs):
    print(f"\n结果 {i+1}:")
    print(f"相似度: {doc.metadata.get('score', 'N/A')}")
    print(f"内容: {doc.page_content[:300]}...")

3. 记忆管理

会话记忆示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 初始化记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    k=5  # 保留最近5轮对话
)

# 初始化对话链
llm = ChatOpenAI(model="gpt-3.5-turbo")
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 多轮对话
response = conversation.predict(input="你好，我是小红")
print(response)

response = conversation.predict(input="我喜欢机器学习")
print(response)

response = conversation.predict(input="我刚才提到了什么？")
print(response)

高级记忆配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain.memory import ConversationSummaryMemory
from langchain.memory import ConversationBufferWindowMemory

# 摘要记忆（适合长对话）
summary_memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

# 窗口记忆（只保留最近N轮）
window_memory = ConversationBufferWindowMemory(
    k=3,  # 只保留最近3轮
    memory_key="chat_history",
    return_messages=True
)

4. 链操作

简单链示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业的翻译助手"),
    ("human", "请将以下文本翻译成{target_language}：\n\n{text}")
])

# 创建链
chain = LLMChain(
    llm=llm,
    prompt=prompt,
    verbose=True
)

# 执行链
result = chain.run({
    "target_language": "英文",
    "text": "我爱自然语言处理"
})
print(result)

顺序链示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from langchain.chains import SequentialChain

# 第一个链：翻译
translate_chain = LLMChain(
    llm=llm,
    prompt=ChatPromptTemplate.from_messages([
        ("system", "你是翻译专家"),
        ("human", "将'{text}'翻译成英文")
    ]),
    output_key="translated_text"
)

# 第二个链：总结
summarize_chain = LLMChain(
    llm=llm,
    prompt=ChatPromptTemplate.from_messages([
        ("system", "你是总结专家"),
        ("human", "总结以下文本：'{translated_text}'")
    ]),
    output_key="summary"
)

# 组合成顺序链
overall_chain = SequentialChain(
    chains=[translate_chain, summarize_chain],
    input_variables=["text"],
    output_variables=["translated_text", "summary"],
    verbose=True
)

# 执行
result = overall_chain.run("自然语言处理是人工智能的一个重要分支，它研究如何使计算机能够理解、解释和生成人类语言。")
print(f"翻译结果: {result['translated_text']}")
print(f"总结结果: {result['summary']}")

5. 代理

工具调用代理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import Tool
from langchain.prompts import ChatPromptTemplate
import requests

# 定义工具
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 这里使用模拟数据，实际应用中可以调用真实的天气API
    return f"{city}今天天气晴朗，温度25°C"

def search_web(query: str) -> str:
    """搜索网络获取信息"""
    # 这里使用模拟数据，实际应用中可以调用搜索引擎API
    return f"搜索结果：关于'{query}'的信息..."

tools = [
    Tool(
        name="GetWeather",
        func=get_weather,
        description="获取指定城市的天气信息。参数：city（城市名称）"
    ),
    Tool(
        name="SearchWeb",
        func=search_web,
        description="搜索网络获取最新信息。参数：query（搜索关键词）"
    )
]

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful 的助手，你可以使用工具来回答用户的问题。"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 创建代理
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行代理
response = agent_executor.invoke({
    "input": "今天北京的天气怎么样？明天会下雨吗？"
})
print(response["output"])

实际应用示例

1. 构建聊天机器人

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import MessagesPlaceholder

# 存储会话历史
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的聊天机器人，擅长回答各种问题。"),
    MessagesPlaceholder(variable_name="messages"),
])

# 创建链
chain = prompt | llm

# 添加会话历史
with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="messages",
)

# 运行聊天机器人
config = {"configurable": {"session_id": "user123"}}

response = with_message_history.invoke(
    {"messages": [HumanMessage(content="你好！")]},
    config=config,
)
print(response.content)

response = with_message_history.invoke(
    {"messages": [HumanMessage(content="我刚才说了什么？")]},
    config=config,
)
print(response.content)

2. 构建 RAG 应用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader
from langchain.vectorstores import Chroma

# 加载文档
loader = TextLoader("knowledge_base.txt")
documents = loader.load()

# 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100
)
texts = text_splitter.split_documents(documents)

# 创建向量存储
embeddings = OpenAIEmbeddings()
db = Chroma.from_documents(texts, embeddings)

# 创建检索器
retriever = db.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

# 创建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    verbose=True
)

# 测试 RAG 应用
query = "LangChain 支持哪些向量数据库？"
result = qa_chain({"query": query})

print(f"答案: {result['result']}")
print("\n来源文档:")
for doc in result['source_documents']:
    print(f"- {doc.page_content[:200]}...")

3. 构建智能代理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from langchain.agents import initialize_agent, AgentType
from langchain.tools import WikipediaQueryRun
from langchain.utilities import WikipediaAPIWrapper

# 初始化 Wikipedia 工具
wikipedia_api = WikipediaAPIWrapper()
wikipedia_tool = WikipediaQueryRun(api_wrapper=wikipedia_api)

# 初始化计算器工具
from langchain.tools import CalculatorTool
calculator_tool = CalculatorTool()

# 工具列表
tools = [wikipedia_tool, calculator_tool]

# 初始化代理
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True
)

# 运行代理
response = agent.run("请计算 2024 年的天数，然后告诉我关于 2024 年奥运会的信息。")
print(response)

4. 构建多模态应用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from langchain_community.document_loaders import UnstructuredImageLoader
from langchain_openai import ChatOpenAI

# 加载图片并提取文本
loader = UnstructuredImageLoader("example_image.png")
documents = loader.load()

# 使用多模态模型
llm = ChatOpenAI(model="gpt-4-vision-preview", max_tokens=1024)

# 分析图片内容
response = llm.invoke([
    HumanMessage(
        content=[
            {"type": "text", "text": "请描述这张图片的内容并提取关键信息"},
            {"type": "image_url", "image_url": {"url": "https://example.com/image.jpg"}}
        ]
    )
])

print(response.content)

最佳实践

1. 性能优化

缓存策略

1
2
3
4
5
6
7
8
9
10
from langchain.cache import RedisCache
import redis

# Redis 缓存
r = redis.Redis(host='localhost', port=6379, db=0)
langchain.llm_cache = RedisCache(r)

# 内存缓存
from langchain.cache import InMemoryCache
langchain.llm_cache = InMemoryCache()

异步处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 异步链调用
async def async_chain_demo():
    result = await chain.arun(input="异步处理示例")
    return result

# 并行处理
from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel(
    translation=translate_chain,
    analysis=analysis_chain
)

result = parallel_chain.run(text="并行处理示例")

2. 安全考虑

输入验证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def validate_input(input_text: str) -> bool:
    """验证用户输入的安全性"""
    # 检查敏感内容
    sensitive_patterns = ["密码", "账号", "银行卡"]
    for pattern in sensitive_patterns:
        if pattern in input_text:
            return False
    return True

# 使用验证函数
if validate_input(user_input):
    response = chain.run(input=user_input)
else:
    response = "输入包含敏感信息，请重新输入。"

输出过滤

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

class SafeResponse(BaseModel):
    """安全的响应格式"""
    content: str = Field(description="响应内容")
    is_safe: bool = Field(description="是否安全")

parser = PydanticOutputParser(pydantic_object=SafeResponse)

# 安全提示
safe_prompt = ChatPromptTemplate.from_messages([
    ("system", "请确保你的回答安全、合规，不包含有害内容。"),
    ("human", "{input}\n\n{format_instructions}"),
])

chain = safe_prompt | llm | parser

3. 监控和调试

LangSmith 集成

1
2
3
4
5
6
7
8
9
10
11
# 启用 LangSmith 跟踪
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "your-project"

# 自定义回调
from langchain.callbacks import FileCallbackHandler

handler = FileCallbackHandler("chain_logs.txt")
chain.run(input="测试日志", callbacks=[handler])

4. 部署考虑

FastAPI 部署

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI
from langchain.serve import add_routes

app = FastAPI()

# 添加 LangChain 路由
add_routes(
    app,
    chain,
    path="/chat",
    enabled_endpoints=["invoke", "batch", "stream"]
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Docker 部署

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

ENV PYTHONUNBUFFERED=1
ENV LANGCHAIN_TRACING_V2=true

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

常见问题

1. 模型调用失败

问题：API 调用超时或返回错误
解决方案：

  • 检查 API 密钥是否正确

  • 验证网络连接

  • 增加超时时间

  • 实现重试机制

1
2
3
4
5
6
7
8
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(prompt):
    return llm.invoke(prompt)

2. 内存使用过高

问题：处理大量文档时内存不足
解决方案：

  • 分批处理文档

  • 使用更高效的文本分割策略

  • 选择合适的向量数据库

1
2
3
4
5
# 分批处理
batch_size = 100
for i in range(0, len(documents), batch_size):
    batch = documents[i:i+batch_size]
    vectorstore.add_documents(batch)

3. 响应时间过长

问题：API 调用响应时间长
解决方案：

  • 使用流式输出

  • 优化提示词

  • 减少上下文长度

  • 使用缓存

1
2
3
# 流式输出
for chunk in chain.stream(input="流式输出示例"):
    print(chunk.content, end="", flush=True)

4. 成本控制

问题：API 调用成本过高
解决方案：

  • 合理设置温度参数

  • 控制最大 tokens

  • 使用缓存减少重复调用

  • 选择合适的模型

1
2
3
4
5
6
7
# 成本优化配置
llm = ChatOpenAI(
    model="gpt-3.5-turbo",  # 选择更经济的模型
    temperature=0.3,        # 降低温度减少随机性
    max_tokens=512,         # 限制输出长度
    request_timeout=10      # 设置超时时间
)

参考资源

官方文档

社区资源

学习资源

相关项目

  • LangGraph：构建图结构的复杂应用

  • LangServe：部署 LangChain 应用

  • LangSmith：监控和评估 LLM 应用