LangChain 完全指南：构建下一代 LLM 应用的核心框架

前言

2023年，随着大语言模型（LLM）的爆发式发展，如何将 LLM 能力集成到实际应用中成为开发者面临的核心挑战。LangChain 应运而生，迅速成为 AI 应用开发领域最热门的框架之一。

LangChain 是一个开源框架，专门用于简化由大语言模型驱动的应用程序的开发。它提供了一套标准化的接口和工具，让开发者能够轻松构建从简单的问答系统到复杂的多步骤 AI Agent。

本文将从 LangChain 的核心概念、架构设计、关键组件、实战案例等多个维度，全面解析这个革命性的 AI 应用开发框架。

一、LangChain 概述

1.1 什么是 LangChain

LangChain 是由 Harrison Chase 创建的开源项目，旨在为大语言模型应用开发提供标准化接口。它最初针对 Python 和 JavaScript 两种语言，现在已经发展成为一个庞大的生态系统。

LLM 应用

LangChain

Models I/O

Chains

Agents

Memory

Tools

1.2 核心设计理念

设计原则 实现方式
模块化 每个组件可独立使用和组合
可组合性 Chains 和 Agents 的链式调用
标准化 统一的接口抽象不同模型
可扩展性 支持自定义组件和集成

二、LangChain 核心组件

2.1 Models I/O

LangChain 提供了与各种 LLM 交互的标准化接口：

from langchain_openai import ChatOpenAI
from langchain_community.llms import HuggingFacePipeline

# OpenAI 模型
llm = ChatOpenAI(
    model="gpt-4",
    temperature=0.7,
    api_key="your-api-key"
)

# Hugging Face 模型
hf_llm = HuggingFacePipeline.from_model_id(
    model_id="google/flan-t5-large",
    task="text2text-generation"
)

# 统一调用接口
response = llm.invoke("什么是 LangChain?")
print(response.content)

支持的模型提供商：

  • OpenAI (GPT-4, GPT-3.5)
  • Anthropic (Claude)
  • Google (Gemini, PaLM)
  • Hugging Face (开源模型)
  • 本地模型 (Ollama, LLaMA)

2.2 Prompts

提示工程是 LLM 应用的核心。LangChain 提供了强大的提示管理功能：

from langchain_core.prompts import PromptTemplate
from langchain_core.prompts import ChatPromptTemplate

# 基础提示模板
prompt = PromptTemplate(
    input_variables=["product", "audience"],
    template="为{audience}写一篇关于{product}的营销文案。"
)

# 对话提示模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的技术文档作者。"),
    ("human", "请解释 {topic} 的核心概念。"),
    ("human", "{followup_question}")
])

# 提示组合
few_shot_prompt = ChatPromptTemplate.from_examples(
    examples=[
        {"input": "Python", "output": "Python 是一种高级编程语言..."},
        {"input": "JavaScript", "output": "JavaScript 是一种脚本语言..."}
    ],
    example_prompt=PromptTemplate(
        input_variables=["input", "output"],
        template="输入: {input}\n输出: {output}"
    )
)

2.3 Chains

Chains 是 LangChain 的核心概念，用于将多个组件组合成完整的工作流：

from langchain.chains import LLMChain
from langchain.chains import SimpleSequentialChain
from langchain.chains import SequentialChain

# 基础 Chain
story_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=["plot"],
        template="根据以下情节写一个短篇故事：{plot}"
    )
)

# 顺序 Chain
synopsis_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=["story"],
        template="以下是故事摘要：\n{story}\n\n请用一句话概括这个故事。"
    )
)

sequential_chain = SimpleSequentialChain(
    chains=[story_chain, synopsis_chain],
    verbose=True
)

# 复杂 Chain（多输入输出）
review_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=["synopsis", "reviewer"],
        template="故事摘要：{synopsis}\n\n作为{reviewer}，请评论这个故事。"
    ),
    output_key="review"
)

overall_chain = SequentialChain(
    chains=[story_chain, synopsis_chain, review_chain],
    input_variables=["plot", "reviewer"],
    output_variables=["story", "synopsis", "review"],
    verbose=True
)

2.4 Memory

Memory 组件用于在对话中保存上下文：

from langchain.memory import ConversationBufferMemory
from langchain.memory import ConversationSummaryMemory
from langchain.memory import VectorStoreMemory
from langchain.chains import ConversationChain

# 缓冲内存
buffer_memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 摘要内存
summary_memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_summary"
)

# 向量内存
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
vector_memory = VectorStoreMemory(
    vectorstore=vectorstore,
    memory_key="chat_history"
)

# 使用 Memory
conversation = ConversationChain(
    llm=llm,
    memory=buffer_memory,
    verbose=True
)

conversation.predict(input="你好，我叫小明")
conversation.predict(input="我叫什么名字？")

2.5 Agents

Agents 是 LangChain 最强大的功能，能够使用工具自主完成任务：

from langchain.agents import create_openai_functions_agent
from langchain.agents import AgentExecutor
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper
from langchain_community.utilities import WikipediaAPIWrapper

# 定义工具
search = SerpAPIWrapper()
wiki = WikipediaAPIWrapper()

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于搜索实时信息，输入应为搜索查询"
    ),
    Tool(
        name="Wikipedia",
        func=wiki.run,
        description="用于查找详细定义和背景信息"
    )
]

# 创建 Agent
agent = create_openai_functions_agent(
    llm=llm,
    tools=tools,
    prompt=hub.pull("hwchase17/openai-functions-agent")
)

# 创建 Agent 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5
)

# 执行查询
result = agent_executor.invoke({
    "input": "什么是 LangChain？它有哪些主要特性？"
})

三、LangChain 高级功能

3.1 RAG (检索增强生成)

RAG 是 LangChain 最流行的应用场景：

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_openai import OpenAIEmbeddings

# 1. 加载文档
loader = TextLoader("./documents/my_doc.txt")
documents = loader.load()

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

splits = text_splitter.split_documents(documents)

# 3. 创建向量存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings()
)

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

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

# 查询
query = "LangChain 的核心组件有哪些？"
result = qa_chain({"query": query})

3.2 SQL Agent

让 LLM 能够查询数据库：

from langchain_community.utilities import SQLDatabase
from langchain.chains import create_sql_query_chain
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain.agents import create_openai_sql_agent

# 连接数据库
db = SQLDatabase.from_uri("sqlite:///my_database.db")

# SQL Chain
sql_chain = create_sql_query_chain(
    llm=llm,
    db=db,
    verbose=True
)

# SQL Agent
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
agent = create_openai_sql_agent(
    llm=llm,
    toolkit=toolkit,
    verbose=True
)

# 执行查询
agent_executor.invoke("销售额最高的前5个产品是什么？")

3.3 自定义工具

创建自己的工具扩展 Agent 能力：

from langchain.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    location: str = Field(description="城市名称")
    unit: str = Field(description="温度单位 (celsius/fahrenheit)")

class WeatherTool(BaseTool):
    name = "weather"
    description = "获取指定城市的当前天气信息"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, location: str, unit: str = "celsius"):
        # 实际调用天气 API
        import requests
        url = f"https://api.weather.com/v1/current?city={location}&unit={unit}"
        response = requests.get(url)
        return response.json()

# 使用自定义工具
weather_tool = WeatherTool()
agent = create_openai_functions_agent(
    llm=llm,
    tools=[weather_tool],
    prompt=hub.pull("hwchase17/openai-functions-agent")
)

四、LangChain 表达式语言 (LCEL)

4.1 LCEL 基础

LCEL (LangChain Expression Language) 是声明式定义 Chain 的新方式：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 传统 Chain 方式
prompt = ChatPromptTemplate.from_template("告诉我关于 {topic} 的趣事")
chain = LLMChain(llm=llm, prompt=prompt)

# LCEL 方式
chain = prompt | llm | StrOutputParser()

result = chain.invoke({"topic": "Python"})

4.2 LCEL 高级用法

from langchain_core.runnables import RunnableBranch
from langchain_core.runnables import RunnableLambda

# 并行执行
from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    summary = (
        {"context": retriever} |
        PromptTemplate.from_template(
            "上下文：{context}\n\n问题：{question}"
        ) | llm | StrOutputParser()
    ),
    sources = (
        {"context": retriever} |
        (lambda x: "\n".join([doc.metadata["source"] for doc in x["context"]]))
    )
)

# 条件分支
branch = RunnableBranch(
    (lambda x: x["language"] == "python", python_chain),
    (lambda x: x["language"] == "javascript", javascript_chain),
    default_chain
)

full_chain = {
    "language": itemgetter("language"),
    "question": itemgetter("question")
} | branch

五、实战案例：构建智能问答系统

5.1 需求分析

构建一个能够回答关于技术文档问题的智能问答系统：

  • 支持多种文档格式（PDF、TXT、MD）
  • 支持多轮对话
  • 能够引用来源
  • 支持实时信息查询

5.2 系统实现

from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 初始化
llm = ChatOpenAI(model="gpt-4", temperature=0)
embeddings = OpenAIEmbeddings()

# 加载文档
def load_documents(folder_path):
    loaders = {
        ".pdf": PyPDFLoader,
        ".txt": TextLoader,
        ".md": TextLoader
    }

    documents = []
    for file_path in Path(folder_path).rglob("*"):
        ext = file_path.suffix
        if ext in loaders:
            loader = loaders[ext](str(file_path))
            documents.extend(loader.load())

    return documents

# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)

# 创建向量存储
def create_vectorstore(documents):
    splits = text_splitter.split_documents(documents)
    vectorstore = Chroma.from_documents(
        documents=splits,
        embedding=embeddings,
        persist_directory="./chroma_db"
    )
    return vectorstore

# 创建 QA Chain
def create_qa_chain(vectorstore):
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True
    )

    qa_chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        return_source_documents=True,
        verbose=True
    )

    return qa_chain

# 使用系统
documents = load_documents("./docs/")
vectorstore = create_vectorstore(documents)
qa_chain = create_qa_chain(vectorstore)

# 对话
while True:
    query = input("\n请输入问题（输入 'quit' 退出）：")
    if query.lower() == 'quit':
        break

    result = qa_chain({"question": query})
    print(f"\n答案：{result['answer']}")
    print(f"\n来源：{[doc.metadata['source'] for doc in result['source_documents']]}")

六、LangChain 生态

6.1 LangSmith

LangSmith 是 LangChain 的官方开发者平台：

功能 描述
调试 可视化 Chain 执行过程
测试 自动化测试和评估
监控 生产环境性能监控
版本管理 Prompt 和 Chain 版本控制 
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# Chain 会自动被追踪
chain.invoke({"input": "test"})

6.2 LangServe

LangChain 应用的生产部署框架：

from fastapi import FastAPI
from langchain_openai import ChatOpenAI
from langserve import add_routes

app = FastAPI()

llm = ChatOpenAI()
chain = llm | StrOutputParser()

# 添加路由
add_routes(
    app,
    chain,
    path="/chat"
)

# 启动服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

七、最佳实践与性能优化

7.1 性能优化

优化策略 实现方法
批量处理 使用 batch() 并行处理多个输入
异步调用 使用 ainvoke() 和 abatch()
流式输出 使用 stream() 实现实时响应
缓存机制 启用语义缓存减少重复调用 
# 批量处理
results = chain.batch([
    {"input": "问题1"},
    {"input": "问题2"},
    {"input": "问题3"}
])

# 异步调用
result = await chain.ainvoke({"input": "异步问题"})

# 流式输出
for chunk in chain.stream({"input": "流式问题"}):
    print(chunk, end="", flush=True)

# 语义缓存
from langchain.cache import InMemoryCache
llm.cache = InMemoryCache()

7.2 错误处理

from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_core.runnables import RunnableConfig
import time

def func_with_timeout(input_data: str):
    time.sleep(2)
    return f"处理结果: {input_data}"

# 添加超时和重试
from langchain_core.runnables import RunnableLambda
from langchain.retries import Retry retry = Retry(
    max_attempts=3,
    stop_after_attempt=10
)

chain = RunnableLambda(func_with_timeout) | retry

八、总结与展望

核心要点

  1. 统一接口：Models I/O 提供标准化的 LLM 交互方式
  2. 可组合性：Chains 和 Agents 实现复杂工作流的灵活组装
  3. 记忆管理：Memory 组件支持多轮对话和上下文保持
  4. 工具生态：丰富的内置工具和自定义工具扩展能力
  5. RAG 应用：检索增强生成是 LangChain 最核心的应用场景

应用场景

  • 智能客服：结合企业知识库的问答系统
  • 文档分析：自动总结、提取、问答
  • 代码助手：结合代码库的编程助手
  • 数据代理：自然语言查询数据库
  • 研究助手：论文阅读和分析

未来发展方向

  • 多模态支持：图像、音频、视频的集成
  • 更强的 Agents：更复杂的规划和推理能力
  • 边缘计算：本地模型部署和优化
  • 企业级功能：权限管理、审计、合规

对于开发者而言，LangChain 不仅是学习 LLM 应用开发的工具，更是参与构建 AI 应用新范式的重要平台。随着 LLM 技术的不断发展，LangChain 将继续演进，成为 AI 应用开发的基础设施。

参考资料

本文基于 LangChain 0.1.0 版本撰写，API 可能在新版本中有变化。

Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

Codex 实战：把学习路线变成作品集

avatar AI编程社区
cover

YLB3118@ACP# 8 口 PCIe3.0 转 SATA 高密度存储桥接芯片（对标 ASM1166）

avatar AI编程社区
cover

29.多模态部署：VLM、语音、视频理解

avatar AI编程社区