智泊AI--Langchain（三）

智泊AI–Langchain（三）

 

0.前言

  前面一章节通过代码了解了链的基本概念以及一些相关组件的使用，本章节主要是使用 LangChain 构建 RAG 和工具调用等操作。

 

1.RAG的基本概念

  RAG 的基本概念前面已经有介绍，这里主要是将 LangChain 作为了中间层进行更好的调用与编排。

 

2.文档加载的实现

  在日常需求中，我们处理的知识文档存在很多种类型，比如文本文件、Word 文档、Excel 表格、PDF 文件等。而 LangChain 提供了很强的数据加载能力，从而支持多种插件的数据格式，例如：

  • TXT文档:TextLoader
  • CSV文档:CSVLoader
  • PDF文档:PyPDFLoader
  • JSON文档:JSONLoader
  • HTML文档:UnstructuredHTMLLoader
  • MD文档:UnStructuredMarkdownLoader
  • 文件目录:DirectoryLoader

  加载器有 load 方法，用于从指定的数据源读取数据，将其转换成一个或多个文档。

1
2
3
4
5
6
7
from langchain_community.document_loaders import UnstructuredWordDocumentLoader, Docx2txtLoader

# 1.指定要加载的Word文档路径
loader = Docx2txtLoader("人事管理流程.docx")

# 加载文档、转换格式化成document
documents = loader.load()

 

3.文档切割

  LangChain 也提供了很多现成的分割器：

  • CharacterTextSplitter

  chunk_overlap 表示文本块之间的最大重叠。基于字符（默认以 “\n\n” 为分隔）进行切割，并通过字符数量来测量文本块大小：chunk_size 控制文本块大小，chunk_overlap 控制文本块之间的最大重叠。

  • RecursiveCharacterTextSplitter

  在前面我们已经了解过，除了自然语言，它也支持对特定的编程语言（JavaScript、cpp、go、java、php、python 等等）的代码进行切割。

  • MarkdownHeaderTextSplitter

  可以根据指定的一组标题来切割一个 Markdown 文档。

  • 官方内置分割器地址：

https://reference.langchain.com/python/langchain-text-splitters

1
2
3
4
5
6
7
8
9
10
# 文档切割 递归切割
# separators
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, #切块大小
    chunk_overlap=50,  # 切块重叠大小
    # separators=[".", '\n', '!', '?', ';']
)

# 通过分割器获取document :create_documents   split_documents  传入一个document对象，返回一个document对象列表
split_documents = text_splitter.split_documents(documents)

  LangChain 的文档转换器处理任务中将动作分为了两步，第一步是对文档进行了切割，主要由切割器完成，第二步是将切割后的文档转化为了 Document 对象/数据格式。

  LangChain 中的数据以 Document 对象和向量的形式在各个包装器中流通，该对象包含了文本以及相关元数据，是 LangChain 在 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
#直接了解LangChain中的“文档”(Document)的具体内容，这里我们跳过了文档与文档加载，文档切割和文档转换过程
#文档的模拟数据
#  page_content 存储原始文档内容  metadata（元数据） 存储当前文档描述信息
# documents = [
#     Document(
#         page_content="猫是柔软可爱的动物，但相对独立",
#         metadata={"source": "常见动物宠物文档"},
#     ),
#     Document(
#         page_content="狗是人类很早开始的动物伴侣，具有团队能力",
#         metadata={"source": "常见动物宠物文档"},
#     ),
#     Document(
#         page_content="金鱼是我们常常喂养的观赏动物之一，活泼灵动",
#         metadata={"source": "鱼类宠物文档"},
#     ),
#     Document(
#         page_content="鹦鹉是猛禽，但能够模仿人类的语言",
#         metadata={"source": "飞禽宠物文档"},
#     ),
#     Document(
#         page_content="兔子是小朋友比较喜欢的宠物，但是比较难喂养",
#         metadata={"source": "常见动物宠物文档"},
#     ),
# ]

 

4.嵌入模型包装器

  LangChain 框架提供了一个 Embeddings 的类，为多种文本嵌入模型提供了统一的接口，可以通过文档转换为向量数据，同时将搜索的问题也转换为向量数据。

  实例化的 Embeddings 类则被称为嵌入模型包装器。

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
# 获得一个阿里通义千问嵌入模型的实例，同样在models.py中被包装为get_ali_embeddings()
from langchain_community.embeddings import DashScopeEmbeddings
# 嵌入模型的模型包装器
llm_embeddings = DashScopeEmbeddings(
    # 模型名称
    model=ALI_TONGYI_EMBEDDING_MODEL,
    # API_KEY
    dashscope_api_key=os.getenv(ALI_TONGYI_API_KEY_OS_VAR_NAME)
)

# 用本地ollama的话
# 切换模型只需改 OLLAMA_EMBEDDING_MODEL，可选：bge-m3 / qwen3-embedding
from langchain_ollama import OllamaEmbeddings

OLLAMA_BASE_URL = "http://localhost:11434"
OLLAMA_EMBEDDING_MODEL = "bge-m3"
# 备用：qwen3-embedding（中文效果更好，0.6B）
# OLLAMA_EMBEDDING_MODEL = "qwen3-embedding"

# 嵌入模型的模型包装器
llm_embeddings = OllamaEmbeddings(
    model=OLLAMA_EMBEDDING_MODEL,
    base_url=OLLAMA_BASE_URL,
)

# 实例化向量空间，向量化+向量存储到向量数据库中
vector_store = Chroma.from_documents(documents=split_documents,embedding=llm_embeddings)

 

5.检索器

  检索器是 RAG 系统的核心组件，负责从知识库中查找与用户查询相关的文档片段，返回的对象是 VectorStoreRetriever，除了所展示的四个检索器外还有其他高级的检索器，后续 RAG 深入的时候会提到。

1
2
3
4
5
6
7
8
9
10
11
# 1.创建 FAISS 检索器
retriever_faiss = FAISS.from_texts(texts, embeddings).as_retriever()
# 2.创建 Chroma 检索器
retriever_chroma = Chroma.from_documents(docs, embeddings).as_retriever()
# 3.关键词检索器
retriever_bm25 = BM25Retriever.from_texts(texts)
# 4.混合检索器 (结合语义+关键词)
ensemble_retriever = EnsembleRetriever(
    retrievers=[retriever_faiss, retriever_bm25],
    weights=[0.7, 0.3]  # 权重分配
)

  其中的 as_retriever() 则代表了向量检索，在检索器中，主要分为了三种模式：

  similarity：按照相关性取的前 k 个向量，默认 k=4。

  similarity_score_threshold：设置了相关性阈值，只取高于该阈值的向量。

  mmr：平衡相关性与多样性，主要应用场景：1.摘要任务 2.问答系统 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
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
# 安装 pip install langchain_chroma
# 加载word文档 安装 pip install docx2txt
# 加载json文档 安装 pip install jq
# 加载pdf文档  安装 pip install pymupdf
# 加载HTML文档 安装 pip install unstructured
# 加载MD文档   安装 pip install markdown +  pip install unstructured
import langchain
from langchain_community.vectorstores import Chroma

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.documents import Document
from langchain_core.vectorstores import VectorStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
import os

from models import get_lc_model_client


#获得访问大模型客户端
client = get_lc_model_client()

#直接了解LangChain中的“文档”(Document)的具体内容，这里我们跳过了文档与文档加载，文档切割和文档转换过程
#文档的模拟数据
#  page_content 存储原始文档内容  metadata（元数据） 存储当前文档描述信息
# documents = [
#     Document(
#         page_content="猫是柔软可爱的动物，但相对独立",
#         metadata={"source": "常见动物宠物文档"},
#     ),
#     Document(
#         page_content="狗是人类很早开始的动物伴侣，具有团队能力",
#         metadata={"source": "常见动物宠物文档"},
#     ),
#     Document(
#         page_content="金鱼是我们常常喂养的观赏动物之一，活泼灵动",
#         metadata={"source": "鱼类宠物文档"},
#     ),
#     Document(
#         page_content="鹦鹉是猛禽，但能够模仿人类的语言",
#         metadata={"source": "飞禽宠物文档"},
#     ),
#     Document(
#         page_content="兔子是小朋友比较喜欢的宠物，但是比较难喂养",
#         metadata={"source": "常见动物宠物文档"},
#     ),
# ]

from langchain_community.document_loaders import UnstructuredWordDocumentLoader, Docx2txtLoader

# 1.指定要加载的Word文档路径
# 用 __file__ 定位文档，避免因运行目录不同而找不到文件
DOC_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "人事管理流程.docx")
loader = Docx2txtLoader(DOC_PATH)

# 加载文档、转换格式化成document
documents = loader.load()
# print(len(documents))
# # Document(metadata={'source': '人事管理流程.docx'}, page_content='集团管理制度\n\n人')
# print(documents)

# 文档切割 递归切割
# separators
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, #切块大小
    chunk_overlap=50,  # 切块重叠大小
    # separators=[".", '\n', '!', '?', ';']
)
# 通过分割器获取document :create_documents   split_documents  传入一个document对象，返回一个document对象列表
split_documents = text_splitter.split_documents(documents)

# 模型包装器：大模型分成三类：LLM  聊天模型  嵌入模型
# 这里使用本地 Ollama 的嵌入模型，不需要联网和 API Key
# 切换模型只需改 OLLAMA_EMBEDDING_MODEL，可选：bge-m3 / qwen3-embedding
from langchain_ollama import OllamaEmbeddings

OLLAMA_BASE_URL = "http://localhost:11434"
OLLAMA_EMBEDDING_MODEL = "bge-m3"
# 备用：qwen3-embedding（中文效果更好，0.6B）
# OLLAMA_EMBEDDING_MODEL = "qwen3-embedding"

# 嵌入模型的模型包装器
llm_embeddings = OllamaEmbeddings(
    model=OLLAMA_EMBEDDING_MODEL,
    base_url=OLLAMA_BASE_URL,
)

# 实例化向量空间，向量化+向量存储到向量数据库中
vector_store = Chroma.from_documents(documents=split_documents,embedding=llm_embeddings)


#展示相似度查询，实际业务中可以不要
# print(vector_store.similarity_search("狸花猫"))
# print("--"*15)
# print(vector_store.similarity_search_with_score("狸花猫"))


# print(vector_store.similarity_search("晋升"))
# 从向量数据库检索出来的结果

#按相似度的分数进行排序，分数值越小，越相似（其实是L2距离）
#从向量数据库检索，使用chroma原始API查询   bind(k=1)表示返回相似度最高的第一个 top-K
# print(vector_store.similarity_search_with_score("晋升"))
#
# exit()

# 检索器对象：检索文档，可以根据需要对检索后的结果做各种处理
# VectorStoreRetriever 中allowed_search_types 参数设置检索方式：

# ======"similarity", 默认，向量相似度检索 ======
# retriever = vector_store.as_retriever()

# ======"similarity_score_threshold", 向量相似度阈值检索 ======
# retriever = vector_store.as_retriever(
#     search_type="similarity_score_threshold",
#     search_kwargs={
#         "score_threshold": 0.3,
#     }
# )

# ======"mmr"  max margin relevance :一种平衡相关性和多样性的检索方式======
retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20, "lambda": 0.7}
)

# result = retriever.invoke("晋升")

# print(result)


# exit()

message = """
仅使用提供的上下文回答下面的问题：
{question}
上下文：
{context}
"""
prompt_template = ChatPromptTemplate.from_messages([('human',message)])
# 定义这个链的时候，还不知道问题是什么，
# 用RunnablePassthrough允许我们将用户的具体问题在实际使用过程中进行动态传入
chain = {"question":RunnablePassthrough(),"context":retriever} | prompt_template | client

while True:
    question = input("请输入问题：")
    if question == "exit":
        break
    resp = chain.invoke(question)
    print(resp.content)

 

6.create_retrieval_chain

  前面第二节中，我们提到了 create_retrieval_chain 是用于构建 stuff 文件链的核心函数，这种链的作用是将所有的相关文档直接整合到提示词中，传递给大模型(LLM)生成回答，适用于文档量较小的情况。

  而 create_retrieval_chain 是 LangChain 框架中用于构建检索增强生成系统的核心函数，将文档检索器和文档处理链组合成一个完整的问答流程。

1
2
3
4
5
6
#创建链，预定义链 create_stuff_documents_chain 文档链
chain1 = create_stuff_documents_chain(client,prompt_template)
#  参数1:是检索器  参数2:是文档链
chain2 = create_retrieval_chain(retriever,chain1)
# 用大模型生成答案
resp = chain2.invoke({"input":"会议说了什么?"})

 

7.工具调用

  大模型需要使用工具的前提是必须找到各种工具的使用说明，当遇到具体场景，能够从工具库中选择合适的工具使用。

7.1 自定义工具

  在 LangChain 语法中使用 @tool 定义工具，同时需要加入文本的描述，让大模型理解工具的作用，需要注意的是工具函数的描述一定要在第一行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langchain.tools import tool
from langchain.agents import create_agent

@tool
def get_date():
    """ 获取今天的具体日期 """
    return datetime.date.today().strftime("%Y-%m-%d")

# 大模型客户端绑定工具
agent = create_agent(
    model,
    tools=[get_date, open_browser],
    #system_prompt="你是一个专业的浏览器助手，你可以打开网站，也可以获取日期",
    #checkpoint=True,
)

  也可以通过 Tool 自定义方法。

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
from langchain.tools import tool
from langchain.agents import create_agent

tools = [
    # 通过Tool来描述声明工具 name 工具名称 func 函数体 description 函数的功能描述
    Tool(
        name="get_current_time", #可以随意，但是建议跟函数名称一致
        func=get_current_time,
        # 函数的功能描述，说明书
        description="当你想知道现在的时间时非常有用。"
    ),
    Tool(
        name="recom_drink",
        func=recom_drink,
        description="用户口渴，为其推荐附近的饮品店"
    ),
    Tool(
        name="open_calc",
        func=open_calc,
        description="打开本地计算机上的计算器。"
    ),
    Tool(
        name="open_browser",
        func=lambda url: open_browser(url),
        description="打开本地计算机上的网页浏览器，并接受网站的url作为参数。"
    )
]


agent = create_agent(
    model=llm,  # 聊天模型
    tools=tools, # 工具列表
    system_prompt=system_prompt,
    checkpointer=memory  # 传入记忆组件
)
7.2 大模型的简单调用

  model.bind_tools([get_date, open_browser]) 使用列表的方式传入，bind_tools 绑定相关工具。

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
import datetime
import os

from dashscope import api_key

from langchain_openai import ChatOpenAI
from langchain.chat_models import  init_chat_model
from langchain_core.prompts import  PromptTemplate
# from langchain_core.tools import tool
from langchain.tools import tool
from models import get_lc_model_client

# api_key  秘钥  model：模型名称 base_url：模型连接的url地址
model = get_lc_model_client()

# resp = model.invoke("今天北京的天气是怎么样的？")
# resp = model.invoke("今天是几月几号？")
# print(resp)
# exit()

@tool
def get_date():
    """ 获取今天的具体日期 """
    return datetime.date.today().strftime("%Y-%m-%d")


import webbrowser
@tool
def open_browser(url, browser_name=None):
    """ 获取浏览器，打开网站 """
    if browser_name:
        # 获取特定浏览器的控制器
        browser = webbrowser.get(browser_name)
    else:
        # 使用默认浏览器
        browser = webbrowser
    # 打开浏览器并导航到指定的URL
    browser.open(url)

# 大模型客户端绑定工具
tool_llm = model.bind_tools([get_date, open_browser])
# tool_llm = model.bind_tools([get_date])
resp = tool_llm.invoke("今天是几月几号？")

# 大模型给的回复，不是结果，而是通过大模型分析，判断哪个工具能解决当前提问（工具选择）
# # 就是通过工具定义时，描述
# print(resp)
# print("---"*20)

all_tools= {
    "get_date": get_date,
    "open_browser": open_browser
}
# 大模型给函数调用的筛选结果，并没有直接调用工具
# 手动执行调用函数的过程
if resp.tool_calls:  # 非空，调用了模型的工具
    for tool_call in resp.tool_calls:
        tool = tool_call["name"]
        print(tool)
        print(tool_call["args"])
        # 从字典中获取工具函数体
        selected_tool = all_tools.get(tool)
        # 手动执行函数
        result = selected_tool.invoke(tool_call["args"])
        print(result)
7.3 agent形式的调用

  agent 的调用在 Langchain 中使用 create_agent 进行包装，在 tools 中填入工具。

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
import datetime
import os

from dashscope import api_key
from langchain.agents import create_agent

from langchain_openai import ChatOpenAI
from langchain.chat_models import  init_chat_model
from langchain_core.prompts import  PromptTemplate
# from langchain_core.tools import tool
from langchain.tools import tool
from models import get_lc_model_client


# api_key  秘钥  model：模型名称 base_url：模型连接的url地址
model = get_lc_model_client()

# 注意：函数的描述必须写在函数体中的第一行
@tool
def get_date():
    """ 获取今天的具体日期 """
    # """ 获取今天的北京的天气 """
    return datetime.date.today().strftime("%Y-%m-%d")


import webbrowser
@tool
def open_browser(url, browser_name=None):
    """ 获取浏览器，打开网站 """
    if browser_name:
        # 获取特定浏览器的控制器
        browser = webbrowser.get(browser_name)
    else:
        # 使用默认浏览器
        browser = webbrowser
    # 打开浏览器并导航到指定的URL
    browser.open(url)

# 大模型客户端绑定工具
agent = create_agent(
    model,
    tools=[get_date, open_browser],
    #system_prompt="你是一个专业的浏览器助手，你可以打开网站，也可以获取日期",
    #checkpoint=True,

)
# 执行agent
# result = agent.invoke({"messages":[{"role":"user","content":"帮我打开淘宝"}]})
# result = agent.invoke({"messages":[{"role":"user","content":"今天是几月几号？"}]})
# 获取北京的今天的天气
result = agent.invoke({"messages":[{"role":"user","content":"帮我打开淘宝"}]})
print( result)
7.4 个人助手实战
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
import gradio as gr
import os
import json
import random
import subprocess
import webbrowser
from datetime import datetime
from http import HTTPStatus
from typing import List, Dict, Any

from langchain.agents import create_agent
from langchain_classic.memory import ConversationBufferWindowMemory
from langchain_core.messages import SystemMessage
from langchain_core.prompts import MessagesPlaceholder, ChatPromptTemplate
from langchain_core.tools import Tool

from models import get_lc_model_client



# 定义工具函数
# 获取当前时间
def get_current_time(input: str = "") -> str:
    """获取当前时间"""
    current_datetime = datetime.now()
    formatted_time = current_datetime.strftime('%Y-%m-%d %H:%M:%S')
    result = f"当前时间：{formatted_time}。"
    print(result)
    return result

# 推荐附近的饮品店
def recom_drink(input: str = "") -> str:
    """推荐附近的饮品店"""
    result = '''距离您500米内有如下饮料店：\n
    1、蜜雪冰城\n
    2、茶颜悦色\n
    另外距离您200米内有惠民便利店，里面应该有矿泉水或其他饮品'''
    return result

# 打开计算器
def open_calc(input: str = "") -> str:
    """打开计算器"""
    try:
        subprocess.Popen(['calc.exe'])
        return "计算器已打开"
    except Exception as e:
        return f"打开计算器失败: {str(e)}"


# 打开浏览器访问指定网址
def open_browser(url: str) -> str:
    """打开浏览器访问指定网址"""
    try:
        webbrowser.open(url)
        return f"已打开浏览器访问 {url}"
    except Exception as e:
        return f"打开浏览器失败: {str(e)}"


# 初始化聊天模型和记忆
llm = get_lc_model_client()
# 创建提示模板
system_prompt = "你是人工智能助手。需要帮助用户解决各种问题。"

# 创建记忆
# 短期记忆构建：智能体中使用InMemorySaver()实现单会话的短期记忆
from langgraph.checkpoint.memory import InMemorySaver
# 创建短期记忆实例
memory = InMemorySaver()

# 创建LangChain工具列表
# 方式二，自定义工具
tools = [
    # 通过Tool来描述声明工具 name 工具名称 func 函数体 description 函数的功能描述
    Tool(
        name="get_current_time", #可以随意，但是建议跟函数名称一致
        func=get_current_time,
        # 函数的功能描述，说明书
        description="当你想知道现在的时间时非常有用。"
    ),
    Tool(
        name="recom_drink",
        func=recom_drink,
        description="用户口渴，为其推荐附近的饮品店"
    ),
    Tool(
        name="open_calc",
        func=open_calc,
        description="打开本地计算机上的计算器。"
    ),
    Tool(
        name="open_browser",
        func=lambda url: open_browser(url),
        description="打开本地计算机上的网页浏览器，并接受网站的url作为参数。"
    )
]


# 这个时候函数工具选择和调用执行，都是通过Agent来完成的
# 创建Agent
agent = create_agent(
    model=llm,  # 聊天模型
    tools=tools, # 工具列表
    system_prompt=system_prompt,
    checkpointer=memory  # 传入记忆组件
)

# 本地测试调用工具
# RAG  调用工具  一个是最新数据大模型不知道     私有的数据大模型不知道
# response = agent.invoke(
#             {"messages": [{"role": "user", "content": "我想用计算器计算？"}]},
#             # 配置会话标识，用于区分不同用户
#             config={"configurable": {"thread_id": "user_1"}}  # 会话唯一标识，用于区分不同用户
#         )
# print(response)
# exit()



# 与前端交互处理LLM响应
def process_llm_response(query, show_history):
    if len(query) == 0:
        return show_history + [("", "")]
    try:
        # 显示用户查询和等待提示
        yield show_history + [(query, "正在查询大模型...")],""

        # 使用Agent处理查询
        response = agent.invoke(
            {"messages": [{"role": "user", "content": query}]},
            # 配置会话标识，用于区分不同用户
            config={"configurable": {"thread_id": "user_1"}}  # 会话唯一标识，用于区分不同用户
        )
        print(f"LLM输出：{response}")

        # 正确提取回答内容 - 只显示output部分
        if isinstance(response, dict) and 'messages' in response:
            response = response["messages"][-1].content
        else:
            # 如果response不是预期的字典格式，尝试提取其他可能的字段
            response = str(response)
            print(f"警告: 响应格式异常: {response}")

        # 返回结果
        yield show_history + [(query, response)],""
    except Exception as e:
        print(f"Error: {e}")
        yield show_history + [(query, "AI助手出错，请重试或者检查")]



# 前端界面展示
with gr.Blocks(title="大模型中Function Call演示") as demo:
    gr.HTML('<center><h1>欢迎大模型中Function Call演示</h1></center>')

    with gr.Row():
        with gr.Column(scale=10):
            chatbot = gr.Chatbot(height=650)  # 空初始值

    with gr.Row():
        msg = gr.Textbox(label="输入", placeholder="您想了解什么呢？")

    with gr.Row():
        examples = gr.Examples(
            examples=[
                '请问如何做红烧牛肉？',
                '料酒可以换成白酒吗？',
                '帮我打开计算器',
                '现在几点了？',
                '帮我访问淘宝网',
                '我渴了'
            ],
            inputs=[msg]
        )

    clear = gr.ClearButton([chatbot, msg])
    msg.submit(process_llm_response, [msg, chatbot], [chatbot,msg])


if __name__ == '__main__':
    demo.launch(server_port=7778)
7.5 预定义工具调用

  在 LangChain 中，除了允许用户自定义 Tools 组件以满足特定的需求外，同时也提供了一组预定义的 Tools 组件，也就是 Toolkits 组件，包含了多个用于实现特定目标的 Tools 组件。

1
from langchain_community.agent_toolkits.load_tools import load_tools

  这里以 Arxiv 为例子，Arxiv 是一个用于检索学术论文的工具，让 Agent 能够搜索和获取到科学文献。

1
2
3
4
from langchain_community.agent_toolkits.load_tools import load_tools

# 导入arxiv工具
tools = load_tools(["arxiv"])

  基于此搭建一个论文查询助手：

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
import os
import re

# 验证模型连接
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage
from models import get_lc_model_client

model = get_lc_model_client()



# 构建一个基于arxiv工具的论文查询智能体，实现根据论文编号查询论文信息的功能。
# 使用load_tools()方法导入arxiv工具：class ArxivQueryRun(BaseTool)
from langchain_community.agent_toolkits.load_tools import load_tools

# 导入arxiv工具
tools = load_tools(["arxiv"])

# 短期记忆构建：使用InMemorySaver()实现单会话的短期记忆
from langgraph.checkpoint.memory import InMemorySaver
# 创建短期记忆实例
memory = InMemorySaver()
# 系统提示词设计：简洁明了地定义Agent的角色和行为准则
system_prompt = "你是一个专业的论文查询助手，使用arxiv工具为用户查询论文信息，回答需简洁准确，包含论文标题、作者、发表时间和核心摘要。"
# 组装并调用Agent
# 使用create_agent()方法组装Agent，并通过invoke()方法调用。
from langchain.agents import create_agent

# 组装Agent
agent = create_agent(
    model=model,
    tools=tools, #添加工具列表，绑定的论文查询的工具
    system_prompt=system_prompt,
    checkpointer=memory  # 传入记忆组件
)

# 调用Agent查询论文
result = agent.invoke(
    {"messages": [{"role": "user", "content": "请查询arxiv论文编号1605.08386的信息"}]},
    # 配置会话标识，用于区分不同用户
    config={"configurable": {"thread_id": "user_1"}}  # 会话唯一标识，用于区分不同用户
)

# 输出结果（取最后一条消息的内容）
print(result["messages"][-1].content)
"""
执行上述代码后，Agent会自动完成以下流程：

接收用户查询，识别需要调用arxiv工具
生成工具调用请求，传入论文编号1605.08386
执行arxiv工具，获取论文信息（标题、作者、发表时间、摘要）
整合工具返回结果，生成自然语言回复
"""
7.6 工具的其他调用方式

  如果场景是确定的，并不需要大模型来决定是否使用工具，直接在链中加入工具即可，这里以数据库查询为例，在链中直接使用 execute_sql_tools 进行操作。

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
93
94
95
96
97
98
99
100
101
102
103
import os
from operator import itemgetter

from langchain_classic.chains.sql_database.query import create_sql_query_chain
from langchain_community.tools import QuerySQLDatabaseTool
from langchain_community.utilities import SQLDatabase

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
import re
from models import get_lc_model_client

#langchain.debug = True

# 使用工具，将自然语言转换成sql,并且执行sql,将查询的结果给模型，然后大模型回复

#获得访问大模型客户端
client = get_lc_model_client()

#数据库配置 pip install mysqlclient      mysqlclient==2.2.7
HOSTNAME ='127.0.0.1'
PORT ='3306'

# 为了方便，进行简单的配置
DATABASE = 'world'
USERNAME = 'root'
PASSWORD ='1234'
MYSQL_URI ='mysql+mysqldb://{}:{}@{}:{}/{}?charset=utf8mb4'.format(USERNAME,PASSWORD,HOSTNAME,PORT,DATABASE)
db = SQLDatabase.from_uri(MYSQL_URI)
# # 获取数据库中所有的表名称
# print(db.get_usable_table_names())
# # 执行sql的函数
# print(db.run('select * from country limit 1'))

# exit()


#2、因为实际产生的sql是形如```sql....```的，无法直接执行，所以需要清理
#自定义一个输出解析器SQLCleaner
class SQLCleaner(StrOutputParser):
    def parse(self, text: str) -> str:
        pattern = r'```sql(.*?)```'
        match = re.search(pattern, text, re.DOTALL)
        if match:
            sql = match.group(1).strip()
            # 某些大模型还会产生类似'SQLQuery:'前缀，必须去除
            sql = re.sub(r'^SQLQuery:', '', sql).strip()
            return sql
        # 某些大模型还会产生类似'SQLQuery:'前缀，必须去除
        text = re.sub(r'^SQLQuery:', '', text).strip()
        return text



# 1.预定义的链，通用的功能，langchain已经将实现流程固定，代码已经定义好
sql_make_chain = create_sql_query_chain(client, db)| SQLCleaner()

# resp = sql_make_chain.invoke({"question":"请从国家表中查询出China的相关数据"})
# resp = sql_make_chain.invoke({"question":"请从城市表中查询出Haag的相关数据"})

# print("实际可用SQL: ",resp)
# print("**"*15)
# exit()

#3、将前面的部分组合起来，得到最终结果
answer_prompt = PromptTemplate.from_template(
    """给定以下用户问题、可能的SQL语句和SQL执行后的结果，回答用户问题
    Question: {question}
    SQL Query: {query}
    SQL Result:{result}
    回答:"""
)
#2.创建一个执行SQL的工具
execute_sql_tools = QuerySQLDatabaseTool(db = db)
# runnable = RunnablePassthrough.assign(query=sql_make_chain)
# print("RunnablePassthrough-1：",runnable.invoke({"question":"请从国家表中查询出China的相关数据"}))
#
# runnable = RunnablePassthrough.assign(query=sql_make_chain)| itemgetter('query')
# print("RunnablePassthrough-2：",runnable.invoke({"question":"请从国家表中查询出China的相关数据"}))
#
# runnable = RunnablePassthrough.assign(query=sql_make_chain)| itemgetter('query') | execute_sql_tools
# print("RunnablePassthrough-3：",runnable.invoke({"question":"请从国家表中查询出China的相关数据"}))
#
# runnable = RunnablePassthrough.assign(query=sql_make_chain).assign(result=itemgetter('query')|execute_sql_tools)
# print("RunnablePassthrough-4：",runnable.invoke({"question":"请从国家表中查询出China的相关数据"}))
#
# exit()
'''通过上面的步骤，就能搞清楚{question}、{query}、{result}这三个字段是如何通过LCEL链一步步获得的
要注意的是result=itemgetter('query')|execute_sql_tools 中，执行顺序是：
itemgetter('query') -> execute_sql_tools -> result=
所以这段代码实际是：result=(itemgetter('query')|execute_sql_tools)'''
chain = (RunnablePassthrough.assign(query=sql_make_chain).assign(result=itemgetter('query')|execute_sql_tools)
         # {question：问题，query：sql , result : 结果}
        |answer_prompt| client| StrOutputParser())


result = chain.invoke(input={"question":"请从国家表中查询出China的相关数据"})
#result = chain.invoke(input={"question":"请问国家表中有多少条数据"})
print("最终执行的结果：",result)
''',如果场景是确定的，并不需要大模型来决定是否使用工具，直接在链中加入工具即可
#但是如果需要大模型来决定是否使用工具，比如场景是动态的或者是以工具组的形式提供工具，那么需要使用Function Call：'''

 

8.中间件系统

  在 langchain 的 1.0 版本中，新增加了中间件系统，用一种细粒度流程控制机制，用在智能体执行过程中拦截、修改或者增强请求与响应的处理逻辑。

  在 before_agent/before_model、after_model/after_agent 中进行处理，从而无需修改核心 Agent 或工具的代码。而中间件系统的作用主要如下：

8.1 内置中间件

  中间件也分为两种，一种是 langchain 提供的内置中间件，例如总摘要中间件 SummarizationMiddleware，使用其对历史对话进行压缩总结。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=llm,
    tools=[],
    checkpointer=memory,
    # 中间件列表，可以多个，多个顺序执行
    middleware=[
        SummarizationMiddleware(
            model=llm,
            max_tokens_before_summary=80,  # 80个token 会触发 摘要总结
            messages_to_keep=1,  # 在总结后保留最后1条消息
            # 可选 summary_prompt=" 可以自定义进行摘要的提示词...",
            summary_prompt="请将以下对话历史进行简洁的摘要，保留关键信息: {messages}"
        ),
    ],
    # 打印Agent执行的过程日志
    debug= True
)

  最终实现：

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
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver

from models import get_lc_model_client
# Middleware 中间件
# 添加中间件的方式：在create_agent
'''
agent = create_agent(
    model=qwen_model,
    tools=[],
    middleware=[SummarizationMiddleware(), HumanInTheLoopMiddleware()],
)
'''

# 内置中间件
#   LangChain 为常见用例提供预构建的中间件：

# SummarizationMiddleware : 总结摘要的中间件
#       当接近会话次数上限时，自动汇总对话历史记录。
#  非常适合：
#     - 持续时间过长的对话超出了上下文窗口。
#     - 多轮对话，历史悠久
#     - 在需要保留完整对话上下文的应用中
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

llm = get_lc_model_client()
# 创建短期记忆实例
memory = InMemorySaver()
agent = create_agent(
    model=llm,
    tools=[],
    checkpointer=memory,
    # 中间件列表，可以多个，多个顺序执行
    middleware=[
        SummarizationMiddleware(
            model=llm,
            max_tokens_before_summary=80,  # 80个token 会触发 摘要总结
            messages_to_keep=1,  # 在总结后保留最后1条消息
            # 可选 summary_prompt=" 可以自定义进行摘要的提示词...",

            # 单一条件:当tokens> = 4000且消息> = 10时触发
            # trigger={"tokens": 4000, "messages": 10},

            # 多重条件-（任意条件必须满足 - 逻辑“或”）。
#             trigger=[
#                 {"tokens": 5000, "messages": 3},
#                 {"tokens": 3000, "messages": 6},
#             ],
            # keep={"messages": 20},

            summary_prompt="请将以下对话历史进行简洁的摘要，保留关键信息: {messages}"
        ),
    ],
    # 打印Agent执行的过程日志
    # debug= True
)

# 模拟长对话触发摘要
print("\n模拟长对话场景...")
demo_messages = [
    "用户询问你是谁",
    "用户计算商品价格：数量10，单价25.5",
    "用户再次询问你能做什么？",
    "用户想要生成一个介绍湖南的文案，要求100字左右，包含三湘四水，人文历史",
    "用户继续询问更多GPU产品信息",
    "用户要求计算2*20"
]

for i, message in enumerate(demo_messages, 1):
    print(f"\n💬 第{i}轮对话: {message}")
    # 循环调用Agent，模拟多轮对话
    result = agent.invoke({
        "messages": [HumanMessage(content=message)]},
        config={"configurable": {"thread_id": "onenull"}}
    )
    print(result["messages"][-1].content)
    # print("执行结果："+result)
8.2 自定义中间件

  自定义中间件取决于我们需要的功能，对应不同的需求，有提供了不同的拦截钩子函数。

  基于装饰器的：适用于单钩中间件的快速简便。

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
import os

from dotenv import load_dotenv
from langchain.agents.middleware import before_model, wrap_model_call
from langchain.agents.middleware import AgentState, ModelRequest, ModelResponse
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.runtime import Runtime
from typing import Any, Callable
from models import get_lc_model_client

# 前置，在调用模型前，执行这个函数
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"即将调用模型： {len(state['messages'])} 个消息")
    return None

# wrap_model_call 环绕，在model调用前后都执行这个函数
@wrap_model_call
def round_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
        print(f"模型调用前置处理 request： request={request}")
        print(f"模型调用前置处理 handler： handler={handler}")
        result = handler(request)  # 调用模型

        print(f"模型调用后，模型返回结果： {result}")
        return result


llm = get_lc_model_client()

agent = create_agent(
    model=llm,
    # 按照列表顺序依次执行中间件
    middleware=[log_before_model, round_model],
)

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "你好",
            }
        ]
    }
)

  基于类的：对于具有多个钩子的复杂中间件来说功能更强大。

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
import os

from dotenv import load_dotenv
from langchain.agents.middleware import before_model, wrap_model_call, AgentMiddleware
from langchain.agents.middleware import AgentState, ModelRequest, ModelResponse
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.runtime import Runtime
from typing import Any, Callable
from models import get_lc_model_client

class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"即将调用模型： {len(state['messages'])} 个消息")
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"模型返回消息: {state['messages'][-1].content}")
        return None

llm = get_lc_model_client()

agent = create_agent(
    model=llm,
    middleware=[LoggingMiddleware()],
)
result = agent.invoke(
    {
        "messages": [{"role": "user","content": "你好",}]
    }
)
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
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import os
from dotenv import load_dotenv
# 验证模型连接
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from models import get_lc_model_client

llm = get_lc_model_client()

# 创建自定义脱敏中间件
from langchain.agents.middleware import AgentMiddleware
from typing import Any, Dict
import re
# 自定义中间件，中间件实现脱敏的，去除电话号码和邮箱信息
class DesensitizeDataMiddleware(AgentMiddleware):
    """脱敏中间件"""

    def __init__(self, patterns: list = None):
        super().__init__()
        self.patterns = patterns or [
            (r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+', '[EMAIL]'),
            (r'(\+86)?1[3-9]\d{9}', '[PHONE]')
        ]

    def _desensitize_text(self, text: str) -> str:
        # 如果内容为空或已经包含脱敏标记，则跳过处理
        if not text or '[EMAIL]' in text or '[PHONE]' in text:
            return text
        # 快速预检查：只有当可能包含敏感信息时才继续处理
        if '@' not in text and not re.search(r'1[3-9]\d{9}', text):
            return text
        print(f"脱敏前: {text}")
        original_text = text
        for pattern, replacement in self.patterns:
            text = re.sub(pattern, replacement, text)
        # 只有当内容发生变化时才打印
        if original_text != text:
            print(f"脱敏后: {text}")
        return text

    def before_model(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """在模型调用前处理"""
        print("中间件DesensitizeDataMiddleware - before_model 被调用")
        if 'messages' in state:
            messages = state['messages']
            processed_any = False
            for message in messages:
                if hasattr(message, 'content') and isinstance(message.content, str):
                    # 只处理非空内容且未被脱敏的内容
                    if message.content and '[EMAIL]' not in message.content and '[PHONE]' not in message.content:
                        # 快速预检查：只有当可能包含敏感信息时才继续处理
                        if '@' in message.content or re.search(r'1[3-9]\d{9}', message.content):
                            # 只有在真正需要处理时才打印日志
                            if not processed_any:
                                print("进行脱敏处理.....")
                            original_content = message.content
                            message.content = self._desensitize_text(message.content)
                            # 只有当内容发生变化时才打印
                            if original_content != message.content:
                                print(f"消息内容已从 '{original_content}' 修改为 '{message.content}'")
                                processed_any = True
            if processed_any:
                print("脱敏处理完成！")
        return state

    def after_model(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """在模型调用后处理"""
        # 只有在真正处理了内容时才打印
        # 这里可以添加后处理逻辑
        return state

# 创建修复后的中间件
desed_middleware = DesensitizeDataMiddleware()

# ===== 测试中间件功能 =====

print("\n" + "=" * 50)
print("开始测试中间件功能")
print("=" * 50)

try:
    # 导入必要的组件
    from langchain_community.agent_toolkits.load_tools import load_tools
    from langgraph.checkpoint.memory import InMemorySaver
    from langchain.agents import create_agent

    # 导入工具，load_tools支持的工具可以在load_tools.py中查看
    tools = load_tools(["arxiv"])

    # 创建短期记忆实例
    memory = InMemorySaver()

    # 系统提示词设计
    system_prompt = "你是一个专业的论文查询助手，使用arxiv工具为用户查询论文信息，回答需简洁准确，包含论文标题、作者、发表时间和核心摘要。"

    # 创建带中间件的Agent
    agent_with_middleware = create_agent(
        model=llm,
        tools=tools,
        system_prompt=system_prompt,
        checkpointer=memory,
        # 加入脱敏的中间件
        middleware=[desed_middleware]
    )

    # 测试---
    # 测试: 包含电子邮件的输入
    print("测试: 电子邮件脱敏")
    email_input = "我的邮箱是test.user@example.com，请帮我查询论文1605.08386"
    print(f"输入内容: {email_input}")

    result1 = agent_with_middleware.invoke(
        {"messages": [{"role": "user", "content": email_input}]},
        config={"configurable": {"thread_id": "middleware_test_1"}}
    )
    print("结果:", result1["messages"][-1].content)

except Exception as e:
    print(f"中间件版本仍然失败: {e}")
智泊AI课程 > Langchain
智泊AI--Langchain（三）
https://one-null-pointer.github.io/2026/09/03/智泊AI--Langchain（三）/
Author
liaoyue
Posted on
September 3, 2026
传送口