LangChain基础

检索器：负责从存储系统中获取相关文档的核心组件，为问答系统、聊天机器人等应用提供知识检索能力。

2201_75762674

1025人浏览 · 2025-09-08 17:22:10

2201_75762674 · 2025-09-08 17:22:10 发布

一、LangChain简介

LangChain是一个用于开发由语言模型驱动的应用程序的框架。内部封装了各种可操作的组件和接口，构建成复杂的应用链。

LangChain提供了组件、链、检索器、代理的工具。

主要组件

Models--加载模型：支持多种语言模型和嵌入模型
Prompts--提示管理：提示管理、优化和序列化
Memory：对话记忆管理
Indexes：文档加载、分割和向量存储
Chains--链式组件：将组件组合成工作流
Agents：让LLM自主决定动作序列

二、简单提及一下RAG

RAG：Retrieval-Augmented Generation，中文意思是检索增强生成。

2.1定义

将大模型内部训练数据与外部知识源（内部文件资料、数据库、网页等）进行结合，将问题与上下文信息结合发送给LLM，避免幻觉问题、知识过时问题、缺乏领域问题。

2.2工作流程

（1）检索：根据用户的提问问题，系统从外部信息中查找相关片段；

（2）增强：将相关片段与用户原始问题进行结合，达到增强效果；

（3）生成：将增强后的提示问题发给大模型处理，让大模型LLM生成回答。

2.3LangChain与RAG之间的关系

LangChain 和 RAG 是绝佳的互补关系。RAG 提供了一个解决LLM核心问题的强大思路，而 LangChain 提供了实现这个思路的最佳实践和高效工具集。 目前，绝大多数基于 LangChain 构建的应用都是 RAG 应用，这使得两者经常被同时提及。

三、LangChain环境

pip install langchain==0.3.27 -i   https://mirrors.aliyun.com/pypi/simple
pip install langchain_community==0.3.27 -i https://mirrors.aliyun.com/pypi/simple/
pip install -U langchain-openai==0.3.28 -i https://mirrors.aliyun.com/pypi/simple/

pip install transformers==4.54.1  -i https://mirrors.aliyun.com/pypi/simple/
pip install langchain-huggingface==0.3.1 -i https://mirrors.aliyun.com/pypi/simple/

#这个建议去官网查看对应的版本
pip install torch==2.5.1 -i https://mirrors.aliyun.com/pypi/simple/

四、Models组件

用于文本生成任务的模型，输入文本、输出文本。

4.1代码调用

from langchain_openai import OpenAI,ChatOpenAI
def Guiji():
    #返回内容比较多
    # llm = OpenAI(
    #     # docker，wsl的key为空
    #     api_key=" 你的API",
    #     model = "Tongyi-Zhiwen/QwenLong-L1-32B",
    #     #模块地址
    #     base_url="https://api.siliconflow.cn/v1",
    #     temperature=0.7
    #     )

    llm = ChatOpenAI(
        # docker，wsl的key为空
        api_key="你的API",
        model = "Tongyi-Zhiwen/QwenLong-L1-32B",
        #模块地址
        base_url="https://api.siliconflow.cn/v1",
        #控制随机性，为0就表示精确
        temperature=0.7
        )

    #invoke:接收一个输入（字符串或消息对象），发送给 LLM，并返回模型的生成结果。
    res = llm.invoke("你好，你是谁？")  
    print(res)

def test1():
    '''
    测试随机性为0时，输出结果是否精准
    '''
    llm = OpenAI(
        # docker，wsl的key为空
        api_key="你的API",
        model = "Tongyi-Zhiwen/QwenLong-L1-32B",
        #模块地址
        base_url="https://api.siliconflow.cn/v1",
        #控制随机性，为0就表示精确，一般默认为0.7
        temperature=0
        )

    #invoke:接收一个输入（字符串或消息对象），发送给 LLM，并返回模型的生成结果。
    res = llm.invoke("四川的省会城市是在哪儿")  
    print(res)

def test2():
    '''
    测试随机性为1-2时，输出结果是否随意
    '''
    llm = OpenAI(
        api_key="你的API",
        model = "Tongyi-Zhiwen/QwenLong-L1-32B",
        #模块地址
        base_url="https://api.siliconflow.cn/v1",
        temperature=1.5
    )
    res = llm.invoke("给我写个笑话")
    print(res)


if __name__ == '__main__':
    # Guiji()
    # test1()
    test2()

4.2参数解释

model:模型选择，填名称

api_key:密钥，自己去注册或者购买

base_url:API根地址

temperature:生成多样性，0-1之间较为稳定、精确，1-2之间随机性提高

max_tokens:生成最大的词数

4.3流失输出stream

代码里面很简单，就是加一个stream=True，还要导入一个模块：

from langchain.callbacks.streaming_stdout import StreamingStOuCallbackHandler

from langchain_openai import OpenAI 
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

def liushi():
    llm = OpenAI(
        api_key="你的API",
        model = "Tongyi-Zhiwen/QwenLong-L1-32B",
        #模块地址
        base_url="https://api.siliconflow.cn/v1",
        #流式输出
        streaming=True,
        #回调函数
        callbacks=[StreamingStdOutCallbackHandler()]
    )
    res = llm.invoke("给我写个笑话,要真实的笑话")


if __name__ == "__main__":
    liushi()

4.4环境变量配置文件

from dotenv import load_dotenv

variable_name = os.getenv("model_name")

load_dotenv()

设置环境变量配置文件，这样的好处在于，统一用一个文件，就算之后要改，只需要改配置文件里面的内容，很省事~

配置文件：.env

env里面的配置：

#env文件必须在根目录
#主要是配置插件
my_url = 123456
model_name = "Tongyi-Zhiwen/QwenLong-L1-32B"
base_model = D:/ai_code/model/Qwen2___5-0___5B-Instruct
OPENAI_API_KEY ="你的API"
OPENAI_API_BASE = "https://api.siliconflow.cn/v1"

from dotenv import load_dotenv
import os
from langchain_openai import OpenAI
#加载环境变量
load_dotenv()
print(os.getenv("my_url"))


llm = OpenAI(model = os.getenv("model_name"))
rs = llm("你好")
print(rs)

4.5聊天模型

Chat Models 基于严格的消息角色系统：这样做的好处是让模型有记忆力的感觉，回答的更准确

消息类型	用途
`SystemMessage`	设定AI行为准则
`HumanMessage`	用户输入内容
`AIMessage`	AI生成回复

代码演示：

from langchain.schema import SystemMessage, HumanMessage,AIMessage

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage, AIMessage
from dotenv import load_dotenv
import os

load_dotenv()

def chat():
    # 创建OpenAI模型
    llm = ChatOpenAI(model=os.getenv("model_name"))
    # 调用模型
    response = llm.invoke([
        SystemMessage(content="你是一个智能助手"),  #系统角色信息
        HumanMessage(content="中秋节的缘由")
    ])
    print(response.content)


def dialogue():
    # 创建OpenAI模型
    llm = ChatOpenAI(model=os.getenv("model_name"))
    # 调用模型
    response = llm.invoke([
        SystemMessage(content="你是一个评论专家"),  #系统角色信息
        HumanMessage(content="评论一下灰太狼"),
        AIMessage(content="灰太狼是一个动漫角色..."),  #模拟系统回复
        HumanMessage(content="它和懒洋洋的关系如何")  #用户问题
    ])
    print(response.content)


if __name__ == '__main__':
    # chat()
    dialogue()

4.6封装类，调用类

封装类的定义是节省重复代码操作，节省时间和内存

封装操作：这个一般另外建一个根文件夹然后写文件代码

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
from transformers import AutoTokenizer,AutoModelForCausalLM,pipeline
from langchain_huggingface import HuggingFacePipeline
import os

# 类封装：减少代码重复

class ChatModel:
    # 初始化函数
    def __init__(self):
        #获取环境变量
        load_dotenv()
    # 获取在线模型
    def get_online_model(self):
        return ChatOpenAI(model = os.getenv("model_name"),temperature=0.7)

    def get_local_model(self):
        # 模型路径
        model_path = os.getenv("base_model")
        # 加载分词器
        tokenizer = AutoTokenizer.from_pretrained(model_path)
        # 加载权重
        model = AutoModelForCausalLM.from_pretrained(model_path)
        # 创建管道
        pipe = pipeline(
            "text-generation", model=model, tokenizer=tokenizer, max_length=512, temperature=0.7
        )
        hf_pipe = HuggingFacePipeline(pipeline=pipe)
        return hf_pipe

调用操作：

from my_chat.my_chat_model import ChatModel

if __name__ == "__main__":
    #调用类
    chat = ChatModel()
    #调用类中的方法函数
    # llm = chat.get_online_model()
    # print((llm.invoke("你好！").content))
    # print("="*20)
    res = chat.get_local_model().invoke("你好吗？")
    print(res)

五、prompts组件

prompt:提示

作用:将用户输入与上下文数据及其他结构化为语言模型能顾理解的格式。

5.1PromptTemplates--提示模版

作用：prompttemplate是prompt模块中最基本的类，允许创建一个带变量的字符串模块，然后运行时用实际值填充这些变量。

代码示例：

简单填充：

from langchain.prompts import PromptTemplate
from my_chat.my_chat_model import ChatModel

def text():
    temple = "你是一个{type}"
    prompt = PromptTemplate(
        input_variables = {"type"},  #输入变量
        template = temple,  #设置模版

    )
    # 填充模板
    res = prompt.format_prompt(type = "翻译")
    print(res.to_string())

if __name__ == "__main__":
    text()

调用模型：prompt | chat

# 推荐系统的模型应用
def one():
    temple = "你是一个{type}，请用韩文翻译你好"
    prompt = PromptTemplate(
        input_variables={"type"},  # 输入变量
        template=temple,  # 设置模版

    )
    # 模型
    chat = ChatModel()
#自定义类里面的方法
    model = chat.get_online_model()
    # 构建链式组件
    chain = prompt | model
    # 提问
    res = chain.invoke({"type":"翻译"})
    print(res.content)

#自行调用

5.2ChatPromptTemplate

聊天提示模版，专门为聊天模型设计：

单次对话：

from langchain.prompts import ChatPromptTemplate
from my_chat.my_chat_model import ChatModel


def chat():
   prompt = ChatPromptTemplate.from_messages(
      [
         ("system","你是一个{singer}"), #定义系统角色
         ("human","请推荐一首韩文歌曲") #定义用户角色
      ]
   )
   chat1 = ChatModel()
   model = chat1.get_online_model()
   res = model.invoke(prompt.format_messages(singer="歌手"))
   print(res)

多轮对话：

def chat2():
   prompt = ChatPromptTemplate.from_messages([
         ("system","你是一个画家"), #定义系统角色
         ("human","请推荐一幅经典画作"), #定义用户角色
         ("ai","我喜欢梵高的星空"),
         ("human","你喜欢莫奈的画吗？")
      ])
   chat = ChatModel()
   model = chat.get_local_model()
   info = prompt.format_prompt()
   res = model.invoke(info)
   print(res)

另一种形式的多轮对话：

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
def text():
   prompt = ChatPromptTemplate.from_messages([
      SystemMessage(content="你是一个英语老师"),
      HumanMessage(content="text"),
      AIMessage(content="好的，请告诉我作文的主题"),
      HumanMessage(content="关于环保")
    ])
   chat = ChatModel()
   model = chat.get_local_model()
   info = prompt.format_prompt(text="请帮我写一篇英语作文")
   res = model.invoke(info)
   print(res)

六、Memory组件

Memory组件主要是用于管理和维护对话或交互历史的核心板块，它使LLM记住之前的交互信息，实现上下文感知诶的对话。

6.1ConversationBufferWindowMemory

翻译：对话缓冲窗口记忆，主要负责保存最近k轮对话

m langchain.memory import ConversationBufferWindowMemory
from langchain_core.messages import HumanMessage
from my_chat.my_chat_model import ChatModel
from langchain.prompts import ChatPromptTemplate
from langchain_core._api import LangChainDeprecationWarning
from langchain.chains import LLMChain
import warnings
# 过滤掉LangChain的弃用警告
warnings.filterwarnings("ignore", category=LangChainDeprecationWarning)

def one():
    #k:存入最近的K条信息，return_messages：返回信息对象
    #input_key:输入问题的键名，output_key:输出问题的键名
    memory = ConversationBufferWindowMemory(k=2,return_messages=True,input_key="input",output_key="output")
    # 保存上下文
    memory.save_context({"input": "你好"}, {"output": "你好！我是AI助手"})
    memory.save_context({"input": "你叫什么名字"}, {"output": "我叫小张"})
    memory.save_context({"input": "今天天气如何？"}, {"output": "今天晴天，25℃"})
    # 查看历史对话信息
    history = memory.load_memory_variables({})
    print(history)
    for i in history["history"]:
        if isinstance(i,HumanMessage):
            print("用户说的：",i.content)
        else:
            print("ai的：", i.content)

def two():

    memory = ConversationBufferWindowMemory(k=2, return_messages=True)
    chat = ChatModel()
    llm = chat.get_local_model()
    prompt = ChatPromptTemplate.from_messages([
        ("system","你是一个礼貌的助手"),
        ("human","{input}")  #定义的变量要和输入的变量一致
    ])
    #创建chat对象，完成自动存储
    chain = LLMChain(
        prompt=prompt,
        llm = llm,
        memory=memory,
        verbose = True   #输出结果，可选参数
    )
    #提问，测试
    res1= chain.invoke({"input":"我叫李四，今年18岁"})
    # print(res1)
    res2 = chain.invoke({"input": "我叫什么"})
    # print(res2)
    res3 = chain.invoke({"input": "我今年多少岁"})
    # print(res3)
    #查看历史对话信息
    history = memory.load_memory_variables({})
    print(history)
    for i in history["history"]:
        if isinstance(i, HumanMessage):
            print("用户说的：", i.content)
        else:
            print("ai说的：", i.content)



if __name__ == "__main__":
    # one()
    two()

6.2RunnableWithMessageHistory

翻译：带有消息历史的可运行组件，复杂会话管理和隔离对话，自动记忆管理。

from langchain.memory import ConversationBufferWindowMemory
from langchain_core.messages import HumanMessage,AIMessage
from langchain_core.runnables.history import RunnableWithMessageHistory
from my_chat.my_chat_model import ChatModel
from langchain.prompts import ChatPromptTemplate,MessagesPlaceholder
from langchain_core._api import LangChainDeprecationWarning
from langchain.chains import LLMChain
import warnings
# 过滤掉LangChain的弃用警告
warnings.filterwarnings("ignore", category=LangChainDeprecationWarning)

# 1. 完全兼容的Memory实现
class CompatibleConversationBufferMemory(ConversationBufferWindowMemory):
    @property
    def messages(self):
            """兼容性补丁：将load_memory_variables的结果转换为messages格式"""
            memory_data = self.load_memory_variables({})
            return memory_data.get(self.memory_key, [])
    def add_messages(self, messages):
        """实现必须的add_messages方法"""
        for message in messages:
            if isinstance(message, HumanMessage):
                self.save_context({"input": message.content}, {"output": ""})
            # elif isinstance(message, AIMessage):
            #     last_input = self.load_memory_variables({})["history"][-2].content
            #     self.save_context({"input": last_input}, {"output": message.content})


def one():
    memory = CompatibleConversationBufferMemory(k=3, return_messages=True)
    chat = ChatModel()
    llm = chat.get_online_model()
    prompt = ChatPromptTemplate.from_messages([
        ("system","你是一个AI助手"),
        MessagesPlaceholder(variable_name="history"),
        ("human","{input}")  #定义的变量要和输入的变量一致
    ])
    #创建chat对象，完成自动存储
    chain = LLMChain(
        prompt=prompt,
        llm = llm,
        memory=memory,
        # verbose = True   #输出结果，可选参数
    )
    #完成记忆功能
    c = RunnableWithMessageHistory(
        chain,
        lambda session_id: memory,  #获取会话id
        input_messages_key="input",   #输入信息的变量名
        history_messages_key="history",  #保存历史信息的变量名

    )
    #开始会话
    res1 = c.invoke({"input": "我叫李四，今年18岁"},config={"configurable": {"session_id": "test123"}})
    print(res1['text'])
    res2 = c.invoke({"input": "我叫什么"},config={"configurable": {"session_id": "test123"}})
    print(res2['text'])
    res3 = c.invoke({"input": "我今年多少岁"},config={"configurable": {"session_id": "test123"}})
    print(res3['text'])
    # 查看历史对话信息
    history = memory.load_memory_variables({})
    print("完整历史：",history)
    for i in history["history"]:
        if isinstance(i, HumanMessage):
            print("用户说的：", i.content)
        else:
            print("ai说的：", i.content)




if __name__ == "__main__":
    one()

6.3小结

CoversationBufferWindowsMemory:负责存储记忆，

RunnableWithMessageHistory:将带有历史记忆信息的组件运行

七、Indexes组件

7.1index组件所需安装包

pip install langchain-chroma -i https://mirrors.aliyun.com/pypi/simple/

pip install pypdf  -i https://mirrors.aliyun.com/pypi/simple/ 

pip install beautifulsoup4 -i https://mirrors.aliyun.com/pypi/simple/

pip install docx2txt -i https://mirrors.aliyun.com/pypi/simple/

pip install langchain-chroma -i https://mirrors.aliyun.com/pypi/simple/

pip install -U dashscope -i https://mirrors.aliyun.com/pypi/simple/

7.2文档加载

实现多种文档的加载，包括文本文档txt、pdf文档、网页文档HTML、csv文档、word文档

from  langchain_community.document_loaders import TextLoader,PyPDFLoader,WebBaseLoader,CSVLoader,Docx2txtLoader

def txt():
    loader = TextLoader('./data/data/example.txt',encoding="utf-8")
    docs = loader.load()
    print(docs)
    for doc in docs:
        print(doc.page_content)

def pdf():
    loader = PyPDFLoader('./data/data/example.pdf')
    docs = loader.load()
    print(docs)
    for doc in docs:
        print(doc.page_content)


def html():
    loader = WebBaseLoader('https://www.gaokao.com/e/20250604/684012cf1c697.shtml',encoding="gb2312")
    docs = loader.load_and_split()
    print(docs)
    for doc in docs:
        print(doc.page_content)

def csv():
    loader = CSVLoader('./data/data/example.csv', encoding="utf-8")
    docs = loader.load_and_split()
    print(docs)
    for doc in docs:
        print(doc.page_content)

def docx():
    loader = Docx2txtLoader('./data/data/example.docx')
    docs = loader.load_and_split()
    print(docs)
    for doc in docs:
        print(doc.page_content)


if __name__ == "__main__":
    # txt()
    # pdf()
    html()
    # csv()
    # docx()

7.3文本分割器

作用：将长文本分割成合适的片段，使用模型限制、提高精度、保留上下文。

7.3.1字符分割

CharacterTextSplitter，只适合简单应用

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain_text_splitters import RecursiveCharacterTextSplitter


def txt():
    loader = TextLoader('./data/data/example.txt',encoding="utf-8")
    docs = loader.load_and_split()
    print(docs)
    #文本分割
    text = CharacterTextSplitter(
        separator="\n" ,# 分割符
        chunk_size=8,   # 每段分割最大长度
        chunk_overlap=0, # 重叠长度:0---->不重叠

    )
    d = text.split_documents(docs)
    # print(docs)
    for doc in d:
        print("="*30)
        print(doc.page_content)

#递归字符分割
def text():
    loader = TextLoader('./data/data/example.txt',encoding="utf-8")
    docs = loader.load_and_split()
    print(docs)
    #文本分割
    text = RecursiveCharacterTextSplitter(
        # separator=["\n"],
        chunk_size=18,   # 每段分割长度,这是一个可选范围17-26分两端
        chunk_overlap=0, # 重叠长度:0---->不重叠

    )
    d = text.split_documents(docs)
    # print(docs)
    for doc in d:
        print("="*30)
        print(doc.page_content)

#作业分割
def work():
    loader = TextLoader('./data/data/animal.txt',encoding="utf-8")
    docs = loader.load_and_split()
    print(docs)
    #文本分割
    text = RecursiveCharacterTextSplitter(
        # separator=["\n"],
        chunk_size=24,   # 每段分割长度,这是一个可选范围17-26分两端
        chunk_overlap=0, # 重叠长度:0---->不重叠

    )
    d = text.split_documents(docs)
    # print(docs)
    for doc in d:
        print("="*30)
        print(doc.page_content)




if __name__ == "__main__":
    txt()
    # text()
    # work()

7.3.2递归字符分割

RecursiveCharacterTextSplitter，这个更常用

例子在上面

7.4向量存储

自己申请阿里云的密钥：https://help.aliyun.com/zh/model-studio/get-api-key?spm=a2c4g.11186623.help-menu-2400256.d_2_0_0.30d3ded2dq6Nsu&scm=20140722.H_2712195._.OR_help-T_cn~zh-V_1

（1）语义搜索；

（2）相似性搜索；

（3）高效处理数据维度；

（4）多模态支持。

在此基础上，我们要利用文本分割器中的递归字符分割，将长文本分割成我们想要的块，再将这些快存到向量数据库中。

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
#导入向量数据库
from langchain_chroma import Chroma

from my_chat.my_chat_model import ChatModel
import chromadb

def one():
    loader = TextLoader('./data/data/animal.txt',encoding="utf-8")
    docs = loader.load_and_split()
    print(docs)
    # 文本分割
    text = RecursiveCharacterTextSplitter(
        # separator=["\n"],
        chunk_size=24,  # 每段分割长度
        chunk_overlap=0,  # 重叠长度:0---->不重叠

    )
    d = text.split_documents(docs)
    chat = ChatModel()
    embedding = chat.get_embedding_model()
    # 把文档保存到向量数据库
    #d：文档列表，embedding：向量模型，persist_directory：向量数据库的路径，collection_name：数据库名称
    chroma = Chroma.from_documents(d,embedding,persist_directory="../chroma_db",collection_name="animal")
    print("存入成功！")

#查询向量数据库
def two():
    #连接向量数据库
    client = chromadb.PersistentClient(path="../chroma_db")
    #查询集合数据
    collections = client.get_collection(name = "animal")
    print(f"数据的长度为{collections.count()}")
    #获取数据
    data = collections.get()
    print(data)
    for i in range(len(data["ids"])):
        print(f"第{i+1}条数据：")
        print(f"ids:{data["ids"][i]}")
        print(f"数据:{data["documents"][i]}")
        print("__"*20)

#删除指定数据
def three():
    # 连接向量数据库
    client = chromadb.PersistentClient(path="../chroma_db")
    # 查询集合数据
    collections = client.get_collection(name="animal")
    # 删除指定数据
    collections.delete(ids="0efeabf5-b4a2-4733-9b92-994b70265ffc")
    print("删除成功！")


#删除所有数据
def four():
    # 连接向量数据库
    client = chromadb.PersistentClient(path="../chroma_db")
    # 查询集合数据
    collections = client.get_collection(name="animal")
    # 删除数据集合
    client.delete_collection(name="shi")
    print("删除所有数据成功")





if __name__ == "__main__":
    # one()
    two()
    # three()
    # four()

7.5检索器

7.5.1 定义

检索器：负责从存储系统中获取相关文档的核心组件，为问答系统、聊天机器人等应用提供知识检索能力。

7.5.2 步骤

（1）数据切割；

（2）将切割的数据导入向量数据库；

（3）指定相似度的度量方式。

代码演示：

from my_chat.my_chat_model import ChatModel
from langchain_chroma import Chroma

def seek():
    #创建向量模型
    chat = ChatModel()
    emb_model = chat.get_embedding_model()
    #加载已有的集合数据
    store = Chroma(
        persist_directory="../chroma_db",  #存储目录
        embedding_function=emb_model,  #向量模型
        collection_name="animal", #集合名称
        collection_metadata={
            "hnsw:space": "cosine"  #集合的元数据
        }
    )

    #创建一个检索器
    res = store.as_retriever(
        search_type="similarity",search_kwargs={"k":1}
    )
    #检索---通过询问问题
    docs = res.invoke("请解释一下猫")
    print("数据类型：",type(docs))
    print(docs)
    for x in docs:
        print(f"文档来源：{x.metadata['source']}")
        print(f"文档内容:{x.page_content}")


def score():
    # 创建向量模型
    chat = ChatModel()
    emb_model = chat.get_embedding_model()
    # 加载已有的集合数据
    store = Chroma(
        persist_directory="../chroma_db",  # 存储目录
        embedding_function=emb_model,  # 向量模型
        collection_name="animal",  # 集合名称
        collection_metadata={
            "hnsw:space": "cosine"  # 集合的元数据
        }
    )
    #创建查看提问内容和文档的相似度得分
    docs = store.similarity_search_with_score("请解释一下猫",k=2)
    print("查询结果",docs)

    for x,score in docs:
        print(f"文档来源：{x.metadata['source']}")
        print(f"文档内容:{x.page_content}")
        print(f"文档相似度分数:{score}")
        print("-"*30)



if __name__ == "__main__":
    # seek()
    score()

八、Chains组件

链组件，就是一个链接工具：允许你将多个组件（提示模版、模型、工具、其他链等）连接起来，构建复杂的、多步骤的应用程序。

8.1基础链

（1）获取模型

（2）创建提示模版

（3）创建chain链

（4）输出结果

代码演示：

from langchain_core.output_parsers import StrOutputParser
from my_chat.my_chat_model import ChatModel
from langchain.prompts import ChatPromptTemplate

def conversation():
    #获取模型
    chat = ChatModel()
    model = chat.get_online_model()
    #创建一个提示模板
    prompt = ChatPromptTemplate.from_messages(
      [
          ("system","你是一个AI助手"),
          ("human","{question}")
      ]
    )
    #构建prompt
    # prompt = prompt.format_prompt(question="金毛是什么样的")
    #构建一个链,顺序不要写反了
    chain = prompt | model
    res = chain.invoke({"question":"金毛是什么样的"})
    print(res.content)

def output():
    # 获取模型
    chat = ChatModel()
    model = chat.get_online_model()
    # 创建一个提示模板
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", "你是一个AI助手"),
            ("human", "{question}")
        ]
    )
    # 构建prompt
    # prompt = prompt.format_prompt(question="金毛是什么样的")
    # 构建一个链,顺序不要写反了
    chain = prompt | model |StrOutputParser()
    res = chain.invoke({"question": "金毛可爱吗！"})
    print(res)  #得到的结果是一个字符串







if __name__ == '__main__':
    # conversation()
    output()

8.2 输出解析器

8.2.1 StrOutputParser解析器 --输出结果为字符串形式

已经在上一个代码出现了，就是在链chain后面加一个字符输出。

8.2.2 PydanticOutputparser解析器 --将输出结果转化为结构化输出

在函数定义之前，要定义一个输出类的格式，要不然大模型不知道该输出些什么

from pydantic import BaseModel, Field
from my_chat.my_chat_model import ChatModel
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser

#创建一个自定义的数据模型
class Person(BaseModel):
    name:str = Field(description="姓名")
    age:int = Field(description="年龄")
    hobbies:str = Field(description="爱好")
    sex:str = Field(description="性别")

# 创建聊天
def conversation():
    #获取模型
    chat = ChatModel()
    model = chat.get_online_model()
    # 创建自定义的数据模型的解析器
    parse = PydanticOutputParser(pydantic_object=Person)
    #创建一个提示模板
    prompt = ChatPromptTemplate.from_messages(
      [
          ("system","请按照以下格式输出内容：{format_object}"),
          ("human","{question}")
      ]
    )
    #添加解析器到模版内
    prompt = prompt.partial(format_object=parse.get_format_instructions())

    #构建一个链,顺序不要写反了
    chain = prompt | model | parse
    res = chain.invoke({"question":"介绍一下杨洋，包括性别、年龄、爱好"})
    print(res)

if __name__ == "__main__":
    conversation()

8.3 检索问答链--RetrievalQA链

（1）接受用户输入；

（2）根据问题，检索知识库中最相关的片段；

（3）将问题与相关文档组成，形成增强提示，发送给大模型；

（4）大模型根据接受内容回答问题，限制模型结合文本回答内容。

好处：

- **知识更新方便**：只需更新后端的知识库（向量数据库），无需重新训练昂贵的LLM，就能让系统获取最新、最专业的信息。
- **减少幻觉**：要求模型严格基于提供的上下文作答，大大降低了胡编乱造的概率，答案更具可信度。
- **处理长文档**：可以处理远超模型上下文限制的海量文档，因为只需要检索与问题最相关的几个片段

演示代码：

from langchain_core.prompts import ChatPromptTemplate
from my_chat.my_chat_model import ChatModel
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA

def seek():
    #创建向量模型
    chat = ChatModel()
    emb_model = chat.get_embedding_model()
    #加载已有的集合数据
    store = Chroma(
        persist_directory="../chroma_db",  #存储目录
        embedding_function=emb_model,  #向量模型
        collection_name="animal", #集合名称
        collection_metadata={
            "hnsw:space": "cosine"  #集合的元数据
        }
    )

    #创建一个检索器
    res = store.as_retriever(
        search_type="similarity",search_kwargs={"k":2}
    )
    #获取模型
    llm = chat.get_online_model()
    #创建一个问答链
    qa = RetrievalQA.from_chain_type(
        llm = llm,  #模型
        retriever = res,  #检索器
        return_source_documents=True  #返回原文档
    )
    #检索---通过询问问题,用链去问问题
    rs = qa.invoke("请解释一下狗")
    print("数据类型：",type(rs))
    print(rs)
    print("用户问的问题：",rs["query"])
    print("AI返回的答案：",rs["result"])
    for doc in rs["source_documents"]:
        print("原文档来源：",doc.metadata["source"])
        print("原文档内容：", doc.page_content)

#完成RAG的检索增强生成---主要是增强提示
def rag():
    #创建向量模型
    chat = ChatModel()
    emb_model = chat.get_embedding_model()
    #加载已有的集合数据
    store = Chroma(
        persist_directory="../chroma_db",  #存储目录
        embedding_function=emb_model,  #向量模型
        collection_name="animal", #集合名称
        collection_metadata={
            "hnsw:space": "cosine"  #集合的元数据
        }
    )

    #创建一个检索器
    res = store.as_retriever(
        search_type="similarity",search_kwargs={"k":2}
    )
    #获取模型
    llm = chat.get_online_model()
    #增强提示
    prompt = ChatPromptTemplate.from_messages([
        ("system","你是一个动物学专家，请你根据参考文档来回答问题，文档的内容是：{context}"),
        ("human","{question}")
    ])

    #创建一个问答链
    qa = RetrievalQA.from_chain_type(
        chain_type_kwargs={"prompt":prompt},  #增强提示
        llm = llm,
        retriever = res,
        return_source_documents=True , #返回原文档
        input_key="question" #输入的key
    )
    #检索---通过询问问题,用链去问问题
    rs = qa.invoke({"question":"请解释一下狗狗的特性"})
    print("数据类型：",type(rs))
    print(rs)
    print("用户问的问题：",rs["question"])
    print("AI返回的答案：",rs["result"])
    for doc in rs["source_documents"]:
        print("原文档来源：",doc.metadata["source"])
        print("原文档内容：", doc.page_content)

if __name__ == "__main__":
    # seek()
    rag()

九、小结

主要熟悉了LangChain的基本组件，LangChain作为由语言模型驱动的应用程序的框架，通常与rag联用。代码通常是固定的样式，主要是熟悉操作。

2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

智能客服机器人排名靠谱吗？如何找到好的厂商？拆解AI特征，盘点2026头部公司排行与优势

2048 AI社区

互联网又造新词啦——约束工程（Harness Engineering）

本文探讨AI Agent在全栈项目中的核心问题（失控、信任债务、工程范式错位），提出通过"约束工程"框架优化运行环境而非模型本身。关键策略包括上下文工程、架构约束、垃圾回收等闭环机制，并以OpenAI、Anthropic等实验验证其有效性。研究表明：明确的约束能显著提升Agent自主性和代码质量，工程师角色需从编码者转变为环境架构师。但需注意功能验证不足、模型依赖性等潜在问题。