LangChain框架初探-基础及实践篇

前言

LangChain是大语言模型应用快速搭建的框架，目前官网提出了Python和JS版本（Java版本在github上有翻版，对比官方还是差点，后续会单独聊），目前对LangChain的技术博客不算很多，这里通过自己的思考+接合官方文档做一个输出，也是对自己思考的留痕方便后续查漏补缺。

什么是`LangChain`？

在官网上对它用途介绍是:

1	`LangChain simplifies every stage of the LLM application lifecycle`

翻译过来就是说LangChain简化了LLM应用生命周期的每个阶段

对开发大模型的段官网定义有三个，分别是：开发、产品化、部署。这里简单画图帮助理解：

紫色部分是LangChain官方平台:
- LangGraph Platfrom并且提供了基础设施，可用于快速部署Agent应用
- LangSmith更好的跟踪LangChain应用, 提供可观察性、评估及提示工程
黑色和墨绿色部分:
- LangChain模块是核心模块，是所有开发的基石，是所有开发LLM开发的基本抽象，并且支持链式调用
- LangGraph是一个中底级别的编排框架，在LangChain基础上用于构造、管理、部署长时间有状态的代理应用
- IntegrationsLangChain对业务的价值都体现在这，可以快速接入所有主流的大模型以及基础设施层的实现

为什么要学习并使用LangChain？

主要是有几个方面: 认可度、持续维度度、易用性、可扩展性、稳定版本、可观性

首先虽然目前的大模型开发一般都有官方提供的SKD，但是接入没有一个标准，开发起来很混乱，会一直不断的重复造轮子。LangChain的出现解决了这个问题，它大而全的接纳了所有大语言开发，做到了真正意思上的统一，通过几行代码就能实现复杂的AI相关的开发。
其次从自身的角度看，再牛X的程序员也要拥抱变化，AI是大势所趋，学习主流的技术才会永远不被淘汰，紧赶世界的步伐才能不断的成长。
也有人唱衰LangChain，说技术变化如此之快的年代，用 LangChain 来构建一切根本行不通，但是我觉得只要一个框架：
1. 是开源的(star高)
2. 不断有人维护
3. 顺应现在主流
如果有任意两个特点就可以学习，然而明显LangChain三个都满足，所以不用担心是否过时过快。

应用场景列举

智能回答系统(RAG)：基于文档/数据库/网页的智能回答
多轮对话roboot：实现AI的长期记忆、保持用户上下文
智能代理(Agent)：自主完成任务分析和工具调用（人类、AI，主、副驾驶舱模式）
AI智能编辑助手：提供代码TAB, 代码解释等
多模态应用：图文对话、图文生成

二、LangChain核心概念

这里更加聚焦于产品的视角，诠释LangChain的核心概念

Models（模型）：

提供了与各种语言模型（如 OpenAI 的 GPT-3、GPT-4, Hugging Face 的模型等）交互的接口。它支持模型的加载、配置和使用，使得开发者可以轻松地将这些模型集成到自己的应用中。

Chains（链）：

Chains 模块允许开发者构建一系列步骤，这些步骤可以串联起来处理数据。例如，你可以创建一个链，其中包含数据预处理、模型预测、结果解释等步骤。有助于将复杂的任务分解为更小、更易管理的部分

Prompts（提示词）：

这个模块提供了创建和管理提示（prompts）的工具。提示词是输入到语言模型中的预设文本，通过此模块能快速生成提示词用于初始化聊天上下文。

Memory（记忆）：

记忆模块用于在多个请求或对话之间保持上下文。这对于需要持续对话的场景特别有用，例如聊天机器人或虚拟助手。它可以帮助模型记住之前的对话内容，从而提供更加连贯和个性化的响应

Indexes（索引模块）：

Indexes 模块允许开发者将外部数据源（如文档、数据库、知识库等）与 LLM 集成。这通过创建索引来实现，使得 LLM 可以访问并利用这些外部数据源中的信息来生成回答或完成任务

Agents（代理模块）：

Agents 模块允许创建智能体，这些智能体可以自主地与外部世界交互，并根据需要选择合适的工具或模型来执行任务。智能体可以执行复杂的任务，例如搜索信息、调用外部 API、执行复杂的逻辑决策等。

三、LangChain模块

`langchain-core`

该包包含不同组件的基本抽象以及将它们组合在一起的方法。核心组件的接口，如大型语言模型、向量存储、检索器等在此定义。此处未定义任何第三方集成。依赖项故意保持非常轻量级。

`langchain`

主要的 langchain 包含链、代理和检索策略，这些构成了应用程序的认知架构。这些不是第三方集成。这里的所有链、代理和检索策略并不特定于任何一个集成，而是适用于所有集成的通用策略。

`langchain-community`

此包包含由 LangChain 社区维护的第三方集成。关键的合作伙伴包，这包含了各种组件（大型语言模型、向量存储、检索器）的所有集成。此包中的所有依赖项都是可选的，以保持包尽可能轻量。

四、组件使用及技巧

1）Prompt组件及使用技巧

使用langchain_core.prompts下的工具包，可以根据模板生成一个提示词，通过管理模板进行复用提示词

Prompt组件基础用法

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time   : 2025/5/18 12:50
@Author : liual
@File   : 1.Prompt组件基础用法.py
"""
from datetime import datetime

from langchain_core.messages import AIMessage
from langchain_core.prompts import (
    PromptTemplate,
    ChatPromptTemplate,
    HumanMessagePromptTemplate, MessagesPlaceholder)

prompt = PromptTemplate.from_template("请将一个关于{subject}的冷笑话")
print(prompt.format(subject="程序员"))
prompt_value = prompt.invoke({"subject": "程序员"})
print(prompt_value.to_string())
print(prompt_value.to_messages())
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是DeepSeek开发的聊天机器人，请根据用户的提问进行回复，当前时间为：{now}"),
    MessagesPlaceholder("chat_history"),
    HumanMessagePromptTemplate.from_template("请讲一个关于{subject}的冷笑话")
]).partial(now=datetime.now())
chat_prompt_value = chat_prompt.invoke({
    "chat_history": [
        ("human", "我叫刘小帅"),
        AIMessage("你好，我是DeepSeek，有什么可以帮助你")
    ],
    "subject": "程序员"
})
print(chat_prompt_value)
print(chat_prompt_value.to_string())

消息提示词模板拼接

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time   : 2025/5/18 13:29
@Author : liual
@File   : 3. 消息提示词模板拼接.py
"""
from langchain_core.prompts import ChatPromptTemplate

system_chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是深度求索开发的聊天机器人，请根据用户的提问进行回复，我叫{username}")
])

human_chat_prompt = ChatPromptTemplate.from_messages(
    ("human", "{query}")
)

chat_prompt = system_chat_prompt + human_chat_prompt

print(chat_prompt.invoke({
    "username": "DeepSeek",
    "query": "hello, its me"
}))

2）Model组件及使用技巧

Model流式输出

其实可以接入基本上当前市场上的所有主流大模型

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time   : 2025/5/18 14:41
@Author : liual
@File   : 3. Model流式输出.py
"""
import os
from datetime import datetime

import dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是深度求索开发的聊天机器人，请回答用户的问题，现在的时间是{now}"),
    ("human", "{query}"),
]).partial(now=datetime.now)

llm = ChatOpenAI(base_url=os.getenv("OPENAI_URL"), model="deepseek-reasoner")

response = llm.stream(chat_prompt.invoke({"query": "你能简单介绍一下LLM和LLMOps么"}))

for chunk in response:
    print(chunk.content, flush=True, end="")

3）OutPutParser使用技巧

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time   : 2025/5/18 15:46
@Author : liual
@File   : 2.JsonOutPutParser使用技巧.py
"""
import os

import dotenv
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


class Joke(BaseModel):
    joke: str = Field(description="回答用户的冷笑话")
    punchline: str = Field(description="这个笑话的笑点")


dotenv.load_dotenv()
parser = JsonOutputParser(pydantic_object=Joke)

prompt = ChatPromptTemplate.from_template("""请根据用户提问的请求进行回答
{format_instructions}
{query}
""").partial(format_instructions=parser.get_format_instructions())

# print(prompt.format(query="请讲一个程序员的冷笑话"))

llm = ChatOpenAI(base_url=os.getenv("OPENAI_URL"), model="deepseek-reasoner")
joke = parser.invoke(llm.invoke(prompt.invoke({"query": "请讲一个程序员的冷笑话"})))
print(type(joke))
print(joke.get("punchline"))
print(joke)

4）LCEL表达式与Runnable可运行协议

LangChain中引入RunnableSerializable来支持LCEL表达式调用，方便开发，通过chain = prompt | llm | parser 直接调用，解决类回调地狱的问题

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time   : 2025/5/18 17:12
@Author : liual
@File   : 2.LCEL表达式简化版本.py
"""
import os

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

dotenv.load_dotenv()

prompt = ChatPromptTemplate.from_template("{query}")
llm = ChatOpenAI(base_url=os.getenv("OPENAI_URL"), model="deepseek-reasoner")
parser = StrOutputParser()

chain = prompt | llm | parser

ai_message = chain.invoke({"query": "请讲一个程序员的冷笑话"})

for output in ai_message:
    print(output, flush=True, end="")

5）利用回调功能调试链应用-让过程更加透明

LangChain的BaseCallbackHandler提供组件调用生命周期的钩子，方便使用者在特定的实际编制入自己的逻辑，比如日志跟踪、事件处理等等

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time   : 2025/5/18 19:52
@Author : liual
@File   : 1.回调功能使用技巧.py
"""

import os
import time
from typing import Any, Optional, Union
from uuid import UUID

import dotenv
from langchain import ChatOpenAI
from langchain_core.callbacks import StdOutCallbackHandler, BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.outputs import GenerationChunk, ChatGenerationChunk, LLMResult
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

dotenv.load_dotenv()


class LLMOpsCallbackHandler(BaseCallbackHandler):
    start_at: float = 0
    end_at: float = 0

    def on_chat_model_start(
            self,
            serialized: dict[str, Any],
            messages: list[list[BaseMessage]],
            *,
            run_id: UUID,
            parent_run_id: Optional[UUID] = None,
            tags: Optional[list[str]] = None,
            metadata: Optional[dict[str, Any]] = None,
            **kwargs: Any,
    ) -> Any:
        print("聊天模型开始执行了")
        print("serialized: ", serialized)
        print("messages: ", messages)

    def on_llm_new_token(
            self,
            token: str,
            *,
            chunk: Optional[Union[GenerationChunk, ChatGenerationChunk]] = None,
            run_id: UUID,
            parent_run_id: Optional[UUID] = None,
            **kwargs: Any,
    ) -> Any:
        print("token 生成了")
        print("token: ", token)

    def on_llm_start(
            self,
            serialized: dict[str, Any],
            prompts: list[str],
            *,
            run_id: UUID,
            parent_run_id: Optional[UUID] = None,
            tags: Optional[list[str]] = None,
            metadata: Optional[dict[str, Any]] = None,
            **kwargs: Any,
    ) -> Any:
        self.start_at = time.time()

    def on_llm_end(
            self,
            response: LLMResult,
            *,
            run_id: UUID,
            parent_run_id: Optional[UUID] = None,
            **kwargs: Any,
    ) -> Any:
        self.end_at = time.time()
        print(f"llm返回了，共记：{self.end_at - self.start_at}")


prompt = ChatPromptTemplate.from_template("{query}")
llm = ChatOpenAI(base_url=os.getenv("OPENAI_URL"), model="deepseek-chat")
parser = StrOutputParser()

chain = {
            "query": RunnablePassthrough()
        } | prompt | llm | parser

ai_message = chain.invoke("你好，你是谁?", config={"callbacks": [StdOutCallbackHandler(), LLMOpsCallbackHandler()]})

print(ai_message)

6）通过缓冲摘要窗孔记忆

因为大模型每次都是新生成会话，所以如果要连贯的跟大模型对话，要带上历史消息，现在最好的实践就是窗口记忆+丢失消息进行摘要

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time   : 2025/5/22 17:30
@Author : liual
@File   : 1.缓冲窗孔记忆示例.py
"""
import os
from operator import itemgetter

import dotenv
from langchain.memory import ConversationBufferWindowMemory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
llm = ChatOpenAI(base_url=os.getenv("OPENAI_URL"), model="deepseek-chat")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是deepseek开发的聊天机器人，请根据对应的上下文回复用户问题"),
    MessagesPlaceholder("history"),
    ("human", "{query}")
])

memory = ConversationBufferWindowMemory(
    k=2,
    return_messages=True,
    input_key="query",
)

chain = RunnablePassthrough.assign(
    history=RunnableLambda(memory.load_memory_variables) | itemgetter("history"),
) | prompt | llm | StrOutputParser()

while True:
    query = input("Human:")

    if query == "q":
        exit(0)

    chain_input = {"query": query, "history": []}
    response = chain.stream(chain_input)
    print("AI:", flush=True, end="")
    out_put = ""
    for chunk in response:
        out_put += chunk
        print(chunk, flush=True, end="")
    memory.save_context(chain_input, {"output": out_put})
    print("")
    print("history: ", memory.load_memory_variables({}))

AI学习

LangChain框架初探-基础及实践篇

https://liu620.github.io/2025/05/22/LangChain框架初探-基础篇实践篇/

作者

alen

发布于

2025年5月22日

许可协议

redis 8.0新版本发布上一篇

高斯数据库使用手册下一篇

LangChain框架初探-基础及实践篇

前言

什么是LangChain？

为什么要学习并使用LangChain？

应用场景列举

二、LangChain核心概念

Models（模型）：

Chains（链）：

Prompts（提示词）：

Memory（记忆）：

Indexes（索引模块）：

Agents（代理模块）：

三、LangChain模块

langchain-core

langchain

langchain-community

四、组件使用及技巧

1）Prompt组件及使用技巧

Prompt组件基础用法

消息提示词模板拼接

2）Model组件及使用技巧

Model流式输出

3）OutPutParser使用技巧

4）LCEL表达式与Runnable可运行协议

5）利用回调功能调试链应用-让过程更加透明

6）通过缓冲摘要窗孔记忆

什么是`LangChain`？

`langchain-core`

`langchain`

`langchain-community`