一、RAG 与 LangChain：相辅相成的 AI 应用架构

1.1 什么是 RAG？

RAG（检索增强生成）是一种结合了信息检索和生成式 AI 的技术架构，它的核心思想非常直观：

在大型语言模型（LLM）回答问题之前，先从外部知识源（如文档、数据库等）检索相关信息，然后将这些信息作为上下文和问题一起交给 LLM。这样做能有效解决 LLM 的三大痛点：

• 幻觉问题：减少模型编造错误信息的可能性
• 知识过时：让模型能够获取训练数据截止日期之后的新信息
• 领域限制：帮助模型掌握特定领域的专业知识

RAG 的工作流程可以简单概括为三步：检索 → 增强（拼接上下文）→ 生成（LLM）→ 返回答案。

1.2 什么是 LangChain？

LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它提供了一套标准化的组件和接口，让我们能够轻松地将 LLM、外部数据源、逻辑流程等连接起来，构建复杂的应用链。

其核心功能包括：

• 统一的组件接口（支持各种 LLM、向量数据库等）
• 链（Chain）机制，将多个组件按特定顺序组合
• 专门的检索器组件，是实现 RAG 的核心
• 代理（Agent）功能，让 LLM 能自主决定调用哪些工具

1.3 两者的关系

我用一个建筑的类比来理解它们的关系：

• RAG 是 "设计蓝图"，规定了房子应该有哪些房间（检索、增强、生成）以及它们如何连接
• LangChain 是 "建筑工具和预制件"，提供了实现这个蓝图所需的各种工具和组件

简单来说，RAG 告诉我们 "做什么"，而 LangChain 告诉我们 "怎么做"。

二、LangChain 环境搭建

学习 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/

这三个库的作用分别是：

• langchain：核心框架，提供基础组件和接口
• langchain_community：社区贡献的扩展功能
• langchain-openai：专门用于集成 OpenAI API

三、核心组件详解

3.1 Models 组件：与语言模型交互的接口

Models 组件是 LangChain 与各种语言模型交互的核心，它提供了统一的接口，让我们可以轻松切换不同的模型。

3.1.1 初始化 OpenAI 模型

from langchain_openai import OpenAI

def one():
    llm = OpenAI(
        api_key="sk-12hbpwzdifwlxipqrkdfaixmxjxpcxyacfwsoizbkivysolexp12",
        model="Tongyi-Zhiwen/QwenLong-L1-32B",
        base_url="https://api.siliconflow.cn/v1",
        temperature=0.7,
    )

    result = llm.invoke("你好")
    print(result)


def two():
    llm = OpenAI(
        api_key="sk-12hbpwzdifwlxipqrkdfaixmxjxpcxyacfwsoizbkivysolexp12",
        model="Tongyi-Zhiwen/QwenLong-L1-32B",
        base_url="https://api.siliconflow.cn/v1",
        temperature=0.7,
    )
    result = llm.invoke("写一首秋天的诗")
    print(result)

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

其中，temperature参数非常关键，它控制生成文本的随机性：

• 0.0：完全确定性输出，适合需要精确答案的场景
• 0.1~0.5：轻度随机性，适合技术文档生成等
• 0.5~1.0：平衡随机性和合理性，适合通用对话
• 1.0以上：高随机性，适合创意写作或头脑风暴

3.1.2 流式输出

在实际应用中，流式输出能带来更好的用户体验

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

def one():
    llm = OpenAI(
        api_key="sk-12hbpwzdifwlxipqrkdfaixmxjxpcxyacfwsoizbkivysolexp12",
        model="Tongyi-Zhiwen/QwenLong-L1-32B",
        base_url="https://api.siliconflow.cn/v1",
        # temperature=0.7,
        streaming=True,
        callbacks=[StreamingStdOutCallbackHandler()],
    )

    result = llm.invoke("写一个故事")

if __name__ == '__main__':
    one()

3.1.3 使用环境变量管理密钥

为了安全起见，建议使用环境变量管理 API 密钥：

1.创建.env 文件：
​​​

OPENAI_API_KEY=你的密钥
OPENAI_API_BASE=你的API地址
MODEL_NAME=模型名称

2.在代码中加载

from langchain_openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()  # 加载环境变量
llm = OpenAI(model=os.getenv("MODEL_NAME"))

3.1.4 聊天模型 (Chat Models)

聊天模型专为对话场景设计，支持多种消息类型：

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
from langchain.schema import SystemMessage,HumanMessage,AIMessage
import os
load_dotenv()
def one():
    llm  = ChatOpenAI(model=os.getenv("model_name"))
    rs = llm.invoke([
        SystemMessage("你是一个英文翻译"),
        HumanMessage("你好")
    ])
    print(rs.content)


def two():
    llm = ChatOpenAI(model=os.getenv("model_name"))
    rs = llm.invoke([
        SystemMessage("你是一个物理教授"),
        HumanMessage("请你解释一下相对论"),
        SystemMessage("相对论是关于时空和引力的理论..."),
        HumanMessage("这与量子力学有什么不同")
    ])
    print(rs.content)

if __name__ == '__main__':
    one()

3.1.5 加载本地大模型

除了调用 API，LangChain 也支持加载本地模型：

from transformers import AutoModelForCausalLM, AutoTokenizer,pipeline
from langchain_huggingface import HuggingFacePipeline
def get_model():
    # 1 加载本地模型, 并创建分词器和模型
    model_path = "../model/Qwen/Qwen3-0.6B"
    # 加载分词器（将文本转换为模型可理解的 token ID）
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    # 加载模型权重
    model = AutoModelForCausalLM.from_pretrained(model_path)

    # 2 创建 HuggingFace Pipeline
    pipe = pipeline(
        "text-generation",  # 表示文本生成任务
        model=model,
        tokenizer=tokenizer,
        max_new_tokens=512,  # 生成的最大 token 数量（控制输出长度）
        temperature=0.7  # 控制随机性（值越高，输出越多样；值越低，输出越确定）
    )


    # 3 封装为 LangChain LLM
    local_llm = HuggingFacePipeline(pipeline=pipe)
    # 使用 `.invoke()` 调用模型
    # response = local_llm.invoke("请解释一下量子计算的基本原理。")
    # print(response)
    return local_llm

if __name__ == '__main__':
    llm = get_model()
    print(llm.invoke("请写一首诗"))

3.2 Prompts 组件：管理提示词的利器

Prompts 组件帮助我们结构化提示词，使代码更清晰、更易于维护。

3.2.1 PromptTemplate

最基础的提示词模板，适合简单场景：

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

def one():
    temple="你是一个{type}"

    prompt = PromptTemplate(
        input_variables=["type"],
        template=temple,
    )
    rs = prompt.format_prompt(type="翻译")
    print(rs.to_string())


def two():
    temple="你是一个{type},请用英文翻译你好"
    #设置模板的输入变量
    prompt =PromptTemplate(
        input_variables=["type"],#设置输入变量
        template=temple,#设置模板
    )
    #格式化模板
    # rs =prompt.format_prompt(type="翻译"),
    #创建模型
    chat = ChatModel()
    model=chat.get_line_model()
    chain = prompt | model
    rs = chain.invoke({"type":"翻译"})
    print(rs.content)


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

与 LLM 结合使用：

llm = get_local_model()
chain = prompt | llm
print(chain.invoke({"subject": "量子计算"}))

3.2.2 ChatPromptTemplate

专为聊天模型设计，支持多种角色的消息：

from my_chat.my_chat_model import ChatModel
from langchain.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage,HumanMessage,AIMessage

def one():
    prompt = ChatPromptTemplate.from_messages([
        ("system","你是一个物理教授"),# 定义系统角色
        ("human","请你解释一下相对论"),
        ("ai","相对论是什么。。。"),# 定义用户角色
        ("human","这与量子力学有什么不同")
    ])

    chat = ChatModel()
    model = chat.get_line_model()
    info = prompt.format_prompt()
    rs = model.invoke(info)
    print(rs.content)

#chatprompt第二种写法
def two():
    prompt = ChatPromptTemplate.from_messages([
        SystemMessage(content="你是一个物理教授"),
        HumanMessage(content="{text}"),
        # HumanMessagePromptTemplate.from_template("{text}"),
        AIMessage(content="相对论是什么。。。"),
        HumanMessage(content="这与量子力学有什么不同")
    ])
    chat = ChatModel()
    model = chat.get_line_model()
    info = prompt.format_prompt(text="请解释一下相对论是什么")
    rs = model.invoke(info)
    print(rs.content)

if __name__ == '__main__':
    two()

消息角色类型包括：

• "system"：系统消息，设定 AI 行为准则
• "human"：人类消息，用户输入内容
• "ai"：AI 消息，模型的回复

3.3 Memory 组件：管理对话历史

Memory 组件使 LLM 能够记住之前的交互，实现上下文感知的对话。

3.3.1 ConversationBufferWindowMemory

这种记忆方式只保留最近的 k 轮对话：

from langchain.memory import ConversationBufferWindowMemory

# 创建记忆对象
memory = ConversationBufferWindowMemory(
    k=2,  # 保留最近2轮对话
    memory_key="history",
    return_messages=True
)

# 保存上下文
memory.save_context({"input": "你好"}, {"output": "你好！我是AI助手"})
memory.save_context({"input": "我叫张三"}, {"output": "很高兴认识你，张三"})

与链结合使用：

from langchain.chains import LLMChain

llm = get_local_model()
memory = ConversationBufferWindowMemory(k=2)

template = ChatPromptTemplate([
    ("system", "你是一位智能助手"),
    ("human", "{input}"),
])

chain = LLMChain(
    llm=llm,
    prompt=template,
    memory=memory
)

# 多轮对话
chain.invoke({"input": "你好,我叫张三,今年23岁"})
chain.invoke({"input": "我叫什么名字"})  # 模型应该记得名字

3.3.2 RunnableWithMessageHistory

用于管理多会话的记忆，支持不同用户的会话隔离：

from langchain.memory import  ConversationBufferWindowMemory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import HumanMessage,AIMessage
from langchain_core._api import LangChainDeprecationWarning
import warnings
from my_chat.my_chat_model import ChatModel
from langchain.prompts import ChatPromptTemplate,MessagesPlaceholder
from langchain.chains import LLMChain
# 过滤掉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):
            #     print("--------------------")
            #     print(self.load_memory_variables({})["history"])
            #     print(self.load_memory_variables({})["history"])
            #     last_input = self.load_memory_variables({})["history"][-1].content
            #     # print("用户说:", last_input)
            #     # print("AI说:", message.content)
            #     self.save_context({"input": last_input}, {"output": message.content})

def one():
    # 第一步:k =2 存入最近的2条对话，return_messages=True返回消息对象，否则返回字符串
    memory = CompatibleConversationBufferMemory(k=3, return_messages=True)
    # 第二步：创建一个ChatModel对象
    chat = ChatModel()
    llm = chat.get_line_model()
    # 第三步:创建提示模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个AI助手"),
        MessagesPlaceholder(variable_name="history"),#历史对话
        ("human", "{input}"),  # 定义的变量要和memory的输入变量一致

    ])
    # 第四步:创建一个Chain对象，完成自动存储历史对话
    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",#保存历史消息的变量名
    )
    # 第六步：开始会话
    rs1= c.invoke({"input": "我叫张三，今年23岁"}, config={"configurable": {"session_id": "test1231"}})
    print("==",rs1['text'])
    rs2 = c.invoke({"input": "我叫什么"}, config={"configurable": {"session_id": "test1231"}})
    print("==",rs2['text'])
    rs3 = c.invoke({"input": "我今年多大了"}, config={"configurable": {"session_id": "test1231"}})
    print("==",rs3['text'])
    #查看历史对话信息
    history = memory.load_memory_variables({})
    # for i in history["history"]:
    #       if isinstance(i, HumanMessage):
    #            print("用户说:", i.content)
    #       else:
    #            print("AI说:", i.content)

if __name__ == '__main__':
    one()

3.4 Indexes 组件：文档索引与检索

Indexes 组件用于处理文档，为 RAG 应用提供检索能力。

3.4.1 安装向量数据库

我使用 Chroma 作为向量数据库：

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

3.4.2 文档加载器

LangChain 支持多种文档来源：

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


# 加载文本
def one():
    loader = TextLoader("../data/example.txt", encoding="utf-8")
    docs = loader.load()
    print(docs)

def two():
    loader = PyPDFLoader("../data/python.pdf")
    docs = loader.lazy_load()
    for doc in docs:
        print(doc.page_content)

def three():
    loader = WebBaseLoader("https://www.gaokao.com/e/20250604/684012cf1c697.shtml")
    docs = loader.load_and_split()
    print(docs)
    for doc in docs:
        print(doc.page_content)
def four():
    loader = CSVLoader(file_path="../data/example.csv", encoding="utf-8")
    docs = loader.load_and_split()
    print(docs)
    for doc in docs:
        print(doc.page_content)
def five():
    loader = Docx2txtLoader("../data/opencv.docx")
    docs = loader.load_and_split()
    print(docs)
    for doc in docs:
        print(doc.page_content)


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