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的核心概念

image-20250523141838108

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组件基础用法
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
#!/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())
  • 消息提示词模板拼接
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#!/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流式输出

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

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
#!/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使用技巧

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
#!/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 直接调用,解决类回调地狱的问题

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
#!/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提供组件调用生命周期的钩子,方便使用者在特定的实际编制入自己的逻辑,比如日志跟踪、事件处理等等

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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
#!/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)通过缓冲摘要窗孔记忆

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

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
47
48
49
50
51
52
#!/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({}))

LangChain框架初探-基础及实践篇
https://liu620.github.io/2025/05/22/LangChain框架初探-基础篇实践篇/
作者
alen
发布于
2025年5月22日
许可协议