当前位置：首页 > AI > 正文内容

LangChain 开发实战完全指南：从入门到精通

廖万里21小时前AI1

LangChain 是当前最流行的 AI 应用开发框架，它让开发者能够快速构建基于大语言模型的应用。从简单的问答系统到复杂的智能体，LangChain 提供了一套完整的工具链。本文将深入讲解 LangChain 的核心概念、组件使用和实战开发，帮助你掌握 AI 应用开发的核心技能。

一、核心概念

LangChain 是一个开源框架，专门用于开发由大语言模型（LLM）驱动的应用程序。它的设计理念是将 AI 应用的各个组件模块化，通过"链"（Chain）的方式将它们组合起来，形成完整的工作流。

LangChain 的核心价值：

  • 统一接口：支持 OpenAI、Claude、文心一言等多种模型，切换成本低
  • 模块化设计：每个组件职责清晰，易于组合和扩展
  • 开箱即用：内置大量工具和模板，快速原型开发
  • 生产就绪：支持缓存、追踪、评估等生产级特性

二、核心组件详解

2.1 Model I/O：模型的输入输出

Model I/O 是 LangChain 最基础的组件，包含三个核心概念：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 模型（Model）：选择使用哪个 LLM
model = ChatOpenAI(model="gpt-4", temperature=0.7)

# 2. 提示模板（Prompt Template）：动态构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}，请用简洁的语言回答问题。"),
    ("user", "{question}")
])

# 3. 输出解析器（Output Parser）：解析 LLM 的输出
parser = StrOutputParser()

# 组合成链
chain = prompt | model | parser

# 执行
result = chain.invoke({
    "role": "Python 开发工程师",
    "question": "什么是装饰器？"
})
print(result)

2.2 Chains：串联工作流

链是 LangChain 的核心概念，它将多个组件串联成一个完整的处理流程：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

# 创建模型实例
model = ChatOpenAI(model="gpt-4")

# 定义多个处理步骤
# 步骤1：生成文章大纲
outline_prompt = ChatPromptTemplate.from_template(
    "为以下主题生成一个详细的文章大纲：{topic}"
)
outline_chain = outline_prompt | model | StrOutputParser()

# 步骤2：根据大纲生成正文
article_prompt = ChatPromptTemplate.from_template(
    "根据以下大纲，写一篇完整的文章：\n\n{outline}"
)

# 使用 RunnableParallel 并行处理
map_chain = RunnableParallel(
    outline=outline_chain,
    topic=RunnablePassthrough()
)

# 完整链：大纲 -> 正文
full_chain = (
    map_chain 
    | (lambda x: {"outline": x["outline"], "topic": x["topic"]})
    | article_prompt 
    | model 
    | StrOutputParser()
)

# 执行
result = full_chain.invoke("LangChain 入门教程")
print(result)

2.3 Agents：智能体系统

Agent 是 LangChain 最强大的组件，它能让 LLM 自主决定调用哪些工具：

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate

# 定义自定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    weather_data = {
        "北京": "晴天，温度 15°C，空气质量良好",
        "上海": "多云，温度 18°C，有轻微雾霾",
        "广州": "小雨，温度 22°C，湿度较高"
    }
    return weather_data.get(city, f"未找到{city}的天气信息")

@tool
def calculate(expression: str) -> str:
    """执行数学计算，输入数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 创建工具列表
tools = [get_weather, calculate]

# 创建带工具的 Agent
model = ChatOpenAI(model="gpt-4", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以使用工具帮助用户解决问题。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行
result = agent_executor.invoke({
    "input": "北京今天天气怎么样？顺便帮我算一下 25 * 4 + 10"
})
print(result["output"])

2.4 Memory：对话记忆

Memory 组件让应用能够记住上下文，实现连贯的多轮对话：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 创建模型和提示
model = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手，请记住之前的对话内容。"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}")
])

chain = prompt | model | StrOutputParser()

# 创建消息历史存储
message_history = ChatMessageHistory()

# 包装成带记忆的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    lambda session_id: message_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

# 多轮对话测试
print("第一轮：")
response1 = chain_with_history.invoke(
    {"input": "我叫小明，是一名程序员"},
    config={"configurable": {"session_id": "user_001"}}
)
print(response1)

print("\n第二轮：")
response2 = chain_with_history.invoke(
    {"input": "你还记得我叫什么名字吗？我的职业是什么？"},
    config={"configurable": {"session_id": "user_001"}}
)
print(response2)

2.5 Retrieval：检索增强生成（RAG）

RAG 是 LangChain 最受欢迎的应用场景，结合向量数据库实现知识库问答：

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

# 1. 加载文档
loader = TextLoader("./knowledge.txt", encoding="utf-8")
documents = loader.load()

# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "，", " "]
)
splits = text_splitter.split_documents(documents)

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

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

# 5. 构建 RAG 链
model = ChatOpenAI(model="gpt-4")

prompt = ChatPromptTemplate.from_template("""
根据以下上下文回答问题。如果上下文中没有相关信息，请说明"根据已有知识无法回答"。

上下文：
{context}

问题：{question}

请提供详细、准确的回答：
""")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 执行问答
answer = rag_chain.invoke("什么是 LangChain？它有什么优势？")
print(answer)

三、实战案例：构建智能客服系统

下面我们使用 LangChain 构建一个完整的智能客服系统，整合前面学到的所有组件：

"""
LangChain 智能客服系统完整实现
功能：知识库问答 + 多轮对话 + 工具调用
"""
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from typing import Dict, List
import json

# ==================== 配置部分 ====================
class Config:
    MODEL_NAME = "gpt-4"
    TEMPERATURE = 0.3
    CHUNK_SIZE = 500
    CHUNK_OVERLAP = 50
    TOP_K = 3

# ==================== 工具定义 ====================
@tool
def query_order(order_id: str) -> str:
    """查询订单状态，输入订单号"""
    orders = {
        "ORD001": {"status": "已发货", "物流": "顺丰 SF123456", "预计到达": "3月22日"},
        "ORD002": {"status": "配送中", "物流": "京东 JD789012", "预计到达": "今天下午"},
        "ORD003": {"status": "已签收", "签收时间": "3月18日 14:30"}
    }
    if order_id in orders:
        return json.dumps(orders[order_id], ensure_ascii=False)
    return f"未找到订单 {order_id}，请检查订单号是否正确"

@tool
def get_product_info(product_name: str) -> str:
    """查询商品信息，输入商品名称"""
    products = {
        "智能手表": {"价格": "999元", "库存": "充足", "特点": "心率监测、睡眠追踪"},
        "蓝牙耳机": {"价格": "299元", "库存": "紧张", "特点": "降噪、长续航"},
        "充电宝": {"价格": "199元", "库存": "充足", "特点": "20000mAh、快充"}
    }
    if product_name in products:
        return json.dumps(products[product_name], ensure_ascii=False)
    return f"未找到商品 {product_name} 的信息"

@tool  
def submit_complaint(content: str) -> str:
    """提交投诉建议，输入投诉内容"""
    import random
    complaint_id = f"CP{random.randint(10000, 99999)}"
    return f"您的投诉已提交，工单号：{complaint_id}，我们将在24小时内处理"

# ==================== RAG 知识库构建 ====================
def build_knowledge_base(documents: List[str], persist_dir: str = "./kb_chroma"):
    """构建向量知识库"""
    from langchain_text_splitters import RecursiveCharacterTextSplitter
    from langchain_core.documents import Document
    
    docs = [Document(page_content=d) for d in documents]
    
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=Config.CHUNK_SIZE,
        chunk_overlap=Config.CHUNK_OVERLAP
    )
    splits = splitter.split_documents(docs)
    
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma.from_documents(
        documents=splits,
        embedding=embeddings,
        persist_directory=persist_dir
    )
    return vectorstore

# ==================== 智能客服类 ====================
class SmartCustomerService:
    def __init__(self, knowledge_docs: List[str] = None):
        self.model = ChatOpenAI(
            model=Config.MODEL_NAME,
            temperature=Config.TEMPERATURE
        )
        
        if knowledge_docs:
            self.vectorstore = build_knowledge_base(knowledge_docs)
            self.retriever = self.vectorstore.as_retriever(
                search_kwargs={"k": Config.TOP_K}
            )
        else:
            self.retriever = None
        
        self.histories: Dict[str, ChatMessageHistory] = {}
        self.tools = [query_order, get_product_info, submit_complaint]
        self._build_chain()
    
    def _get_history(self, session_id: str) -> ChatMessageHistory:
        if session_id not in self.histories:
            self.histories[session_id] = ChatMessageHistory()
        return self.histories[session_id]
    
    def _build_chain(self):
        system_prompt = """你是一个专业的智能客服助手。

你的能力：
1. 回答关于公司产品、政策的问题
2. 查询订单状态
3. 查询商品信息
4. 处理投诉建议

你的特点：
- 友好、专业、高效
- 记住之前的对话内容
- 无法回答时诚实告知

请根据用户的问题提供帮助。"""

        prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            MessagesPlaceholder("chat_history"),
            ("human", "{input}")
        ])
        
        chain = prompt | self.model | StrOutputParser()
        
        if self.retriever:
            def format_docs(docs):
                return "\n\n".join(d.page_content for d in docs)
            
            chain = (
                {
                    "context": self.retriever | format_docs,
                    "input": RunnablePassthrough(),
                    "chat_history": RunnablePassthrough()
                }
                | ChatPromptTemplate.from_messages([
                    ("system", system_prompt + "\n\n相关知识库内容：\n{context}"),
                    MessagesPlaceholder("chat_history"),
                    ("human", "{input}")
                ])
                | self.model
                | StrOutputParser()
            )
        
        self.chain = RunnableWithMessageHistory(
            chain,
            self._get_history,
            input_messages_key="input",
            history_messages_key="chat_history"
        )
    
    def chat(self, message: str, session_id: str = "default") -> str:
        """与客服对话"""
        response = self.chain.invoke(
            {"input": message},
            config={"configurable": {"session_id": session_id}}
        )
        return response

# ==================== 使用示例 ====================
if __name__ == "__main__":
    knowledge_documents = [
        "退换货政策：商品签收后7天内可无理由退换，需保持商品完好。",
        "配送范围：全国包邮，偏远地区需额外3-5天。",
        "支付方式：支持微信支付、支付宝、银行卡、信用卡分期。",
        "售后服务：主要商品享受一年质保，7x24小时客服在线。",
        "会员权益：银卡会员95折，金卡会员9折，钻石会员85折。"
    ]
    
    service = SmartCustomerService(knowledge_docs=knowledge_documents)
    
    print("=== 智能客服测试 ===\n")
    
    print("用户：退货政策是什么？")
    print(f"客服：{service.chat('退货政策是什么？')}\n")
    
    print("用户：那我刚才问的是什么？")
    print(f"客服：{service.chat('那我刚才问的是什么？')}\n")
    
    print("用户：蓝牙耳机多少钱？")
    print(f"客服：{service.chat('蓝牙耳机多少钱？')}\n")

总结

LangChain 为 AI 应用开发提供了一套完整的解决方案：

  • Model I/O：统一不同模型的接口，降低切换成本
  • Chains：串联多个处理步骤，构建复杂工作流
  • Agents：让 LLM 自主决策，实现真正的智能体
  • Memory：记住对话历史，实现连贯交互
  • Retrieval：结合知识库，突破 LLM 的知识局限

掌握 LangChain，你就能快速构建各类 AI 应用：智能客服、文档问答、代码助手、数据分析 Agent 等。建议从简单的 Chain 开始实践，逐步掌握 Agent 和 RAG 等高级特性。

下一步学习建议：

  1. 阅读 LangChain 官方文档，了解更多内置工具
  2. 尝试不同的向量数据库（Pinecone、Milvus、Weaviate）
  3. 学习 LangSmith 进行调试和监控
  4. 探索 LangGraph 构建更复杂的状态机 Agent

本文链接：https://www.kkkliao.cn/?id=916 转载需授权！

分享到：

版权声明：本文由廖万里的博客发布，如需转载请注明出处。


返回列表

上一篇：GitHub Copilot 使用技巧完全指南：让 AI 成为你最强的编程搭档

下一篇：RAG 知识库搭建实战完全指南：从零构建智能问答系统

“LangChain 开发实战完全指南：从入门到精通” 的相关文章

ChatGPT高级使用技巧大全5天前
AI视频生成Sora使用指南5天前
RAG知识库搭建教程5天前
Python AI 开发必备的 10 个库 - 从入门到精通5天前
Claude 3 完全使用指南 - 比 GPT-4 更强的 AI 助手5天前
RAG 知识库搭建实战 - 让 AI 拥有专属记忆5天前

发表评论

访客

看不清，换一张

◎欢迎参与讨论，请在这里发表您的看法和观点。

© 2022-2026 天桥区万策云网络工作室及济南高新区万策网络工作室提供技术支持
鲁公网安备 37010502001945号
 鲁ICP备2026009861号-1

Powered By Z-BlogPHP. Theme by TOYEAN.