• 提示词优化的相关功能API
  • 调用各类模型的功能API
  • 会话记忆的相关功能API
  • 各类文档管理分析的功能API
  • 构建Agent智能体的相关功能API
  • 各类功能链式执行的能力

pip install langchain langchain-community langchain-ollama dashscope chromadb

  • langchain:核心包
  • Langchain-community:社区支持包提供了更多的第三方模型调用(我们用的阿里云千问模型就需要这个包)
  • Langchain-ollama:0llama支持包，支持调用0llama托管部署的本地模型
  • dashscope:阿里云通义千问的Python SDK
  • chromadb:轻量向量数据库(后续使用)

LLMs大语言模型调用

在线模型调用

# 阿里云大模型调用
from langchain_community.llms.tongyi import Tongyi

# 1 实例化模型
model = Tongyi(model="qwen-plus")  # 注意：环境变量中要有 DASHSCOPE_API_KEY / OPENAI_API_KEY
# 2 模型推理
#res= model.invoke("你是谁？")
#print(res)
# 流式输出
res = model.stream("你是谁？")
for chunk in res:
    print(chunk, end=" ", flush=True)

本地ollama大模型调用

from langchain_ollama import OllamaLLM

# 1 获取模型对象
model = OllamaLLM(
    model="qwen3:0.6b"，
    base_url="http://127.0.0.1:11434",
    temperature=0.1,
    timeout=60
)
# 2 模型推理
#res = model.invoke("你是谁？")
#print(res)
# 流式输出
res = model.stream("你是谁？")
for chunk in res:
    print(chunk, end=" ", flush=True)

Chat聊天模型调用

  • AIMessage:就是AI 输出的消息，可以是针对问题的回答。(0penAI库中的assistant角色）
  • HumanMessage:人类消息就是用户信息，由给出的信息发送给LLMs的提示信息，比如“实现一个快速排序方法”.(0penAI库中的user角色)
  • SystemMessage:可以用于指定模型具体所的环境和背景，如角色扮演等。你可以在这里给个代码专家”，或者"返回json格式”(0penAI库中的system角色)

在线模型调用

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
# 1 获取模型
chat = ChatTongyi(model="qwen3-plus")
# 2 消息
# message = [
#     SystemMessage(content="你是一位唐代的诗人"),
#     AIMessage(content="锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
#     HumanMessage(content="根据你上一首诗的格式，再写一首诗")
# ]
# 消息的另一种写法 （角色，内容） 角色：system/ai/human
message = [
    ("system", "你是一位唐代的诗人"),
    ("ai", "锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
    ("human", "根据你上一首诗的格式，再写一首诗"),
]
res = chat.stream(message)
for chunk in res:
    print(chunk.content, end="", flush=True)

本地ollama大模型调用

from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
# 1 获取模型
chat = ChatOllama(model="qwen3:0.6b")
# 2 消息
# message = [
#     SystemMessage(content="你是一位唐代的诗人"),
#     AIMessage(content="锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
#     HumanMessage(content="根据你上一首诗的格式，再写一首诗")
# ]
# 消息的另一种写法
message = [
    ("system", "你是一位唐代的诗人"),
    ("ai", "锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
    ("human", "根据你上一首诗的格式，再写一首诗"),
]
res = chat.stream(message)
for chunk in res:
    print(chunk.content, end="", flush=True)

Embedding嵌入模型

Embeddings Models嵌入模型的特点:将字符串作为输入，返回一个浮点数的列表(向量)在NLP中，Embeddinq的作用就是将数据进行文本向量化。

在线embedding

from langchain_community.embeddings import DashScopeEmbeddings

# 阿里在线Embedding模型 默认：text-embedding-v1
embed = DashScopeEmbeddings()  # 注意：环境变量中要有 DASHSCOPE_API_KEY
# 转向量
print(embed.embed_query("我喜欢你"))
print(embed.embed_documents(["我喜欢你","我稀饭你","晚上吃啥"]))

本地ollama embedding

from langchain_ollama.embeddings import OllamaEmbeddings

embed = OllamaEmbeddings(
    model="qwen3-embedding:0.6b"  # 需要提前使用ollama安装qwen3-embedding:0.6b
)
print(embed.embed_query("我喜欢你"))
print(embed.embed_documents(["我喜欢你","我稀饭你","晚上吃啥"]))

#-----------------------
from langchain_huggingface import HuggingFaceEmbeddings
embed = HuggingFaceEmbeddings(
        model_name="./bge-small-zh-v1.5",  # 中文推荐模型，需要提前下载bge-small-zh-v1.5
        # model_kwargs={"device": "cuda"},    # 如果有 GPU 可启用
        # cache_folder="./models"             # 可选：指定模型缓存目录
    )
print(embed.embed_query("我喜欢你"))
print(embed.embed_documents(["我喜欢你","我稀饭你","晚上吃啥"]))

prompt

提示词优化在模型应用中非常重要，Lanqchain提供了PromptTemplate类，用来协助优化提示词。
PromptTemplate表示提示词模板，可以构建一个自定义的基础提示词模板，支持变量的注入，最终生成所需的提示词。

PromptTemplate

通用提示词模版，zero-shot思想

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

# zero-shot 思想 （无案例）
prompt_temp = PromptTemplate.from_template(
    "我的邻居性{lastname}, 刚生了{gender}, 帮忙起个名字，简略回答"
)

# 变量注入，生成提示词文本
prompt_text = prompt_temp.format(
    lastname="张", gender="女儿"
)

model = Tongyi(model="qwen-max")
res = model.invoke(input=prompt_text)
print(res)

FewShotPromptTemplate

支持基于模版注入示例信息，few-shot思想

参数:

  • examples:示例数据，list，内套字典
  • example_prompt:示例数据的提示词模板
  • prefix:组装提示词，示例数据前内容
  • suffix:组装提示词，示例数据后内容
  • input_variables:列表，注入的变量列表
from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_community.llms.tongyi import Tongyi
from langchain_ollama import OllamaLLM
# few-shot 思想
# 示例模版
example_prompt = PromptTemplate.from_template(
    "单词：{word}, 反义词：{antonym}"
)
# 示例数据
example_data = [
    {"word": "大", "antonym": "小"},
    {"word": "前", "antonym": "后"}
]
# few-shot 模版
few_shot_prompt_temp = FewShotPromptTemplate(
    example_prompt=example_prompt,                  # 示例数据的模版
    prefix="给出给定词的反义词，有如下示例",               # 示例之前的提示词
    examples=example_data,                          # 示例数据
    suffix="基于示例告诉我，{input_word}的反义词是？",    # 示例之后的提示词
    input_variables=["input_word"]                  # prefix，suffix 中需要注入的变量名
)
prompt_text = few_shot_prompt_temp.invoke(input={"input_word": "左"}).to_string()
print(prompt_text)
model = OllamaLLM(model="qwen3:0.6b")
res = model.invoke(input=prompt_text)
print(res)

ChatPromptTemplate

支持注入任意数量的历史回话信息

通过format_messages方法，从列表中获取多轮回话最为聊天的基础模版

ps：PromptTemplate的form_template只能接入一条消息，ChatPromptTemplate的format_messages可以接入一个list消息

from langchain_core.prompts import ChatPromptTemplate,MessagesPlaceholder
from langchain_ollama import OllamaLLM

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位诗人，可以作诗"),
        MessagesPlaceholder("history"),
        ("human", "非常棒，再来一首诗")
    ]
)

history_data = [
    ("human", "来一首诗"),
    ("ai", "床前明月光，疑似地上霜，举头望明月，低头思故乡"),
    ("human", "好诗，再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦")
]

prompt_value = chat_prompt_template.invoke({"history": history_data})  # 得到的是 StringPromptValue 对象
chat_prompt_text = prompt_value.to_string()  # StringPromptValue 转成 str
print(f"历史对话如下：\n{chat_prompt_text}")

# 调用大模型
model = OllamaLLM(model="qwen3:0.6b")
res = model.invoke(input=chat_prompt_text)
print(res)

format 和 invoke

PromptTemplate, FewShotPromptTemplate, ChatPromptTemplate都有format 和 invoke方法

区别

chain

「将组件串联，上一个组件的输出作为下一个组件的输入」是 Langchain 链(尤其是管道链)的核心工作原理，这也是链式调用的核心价值:实现数据的自动化流转与组件的协同工作，如下:

chain =prompt_template l model
核心前提:即Runnable子类对象才能入链(以及callable、mapping接口子类对象也可加入(用的不多))。

Runnable接口的子类，如下类的继承关系:

基础使用

from langchain_core.prompts import ChatPromptTemplate,MessagesPlaceholder
from langchain_ollama import OllamaLLM
from langchain_community.chat_models.tongyi import ChatTongyi


chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位诗人，可以作诗"),
        MessagesPlaceholder("history"),
        ("human", "非常棒，再来一首诗")
    ]
)

history_data = [
    ("human", "来一首诗"),
    ("ai", "床前明月光，疑似地上霜，举头望明月，低头思故乡"),
    ("human", "好诗，再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦")
]

model = OllamaLLM(model="qwen3:0.6b")
# model = ChatTongyi(model="qwen3-max", api_key="xxx")

# 组成链
chain = chat_prompt_template | model
# chain.invoke或chain.stream 触发链条开始，history_data作为原始输入 -> chat_prompt_template -> model -> 输出
res = chain.invoke({"history": history_data})
print(res)

StrOutputParser

上述链中，要将model输出的内容作为输入再次交给model，但出现错误，因为第一个model输出的是AIMessage类型，model能接受的类型是 PromptValue、str、Sequence,所以报错，这时需要用到StrOutputparser

Str0utputParser是LangChain内置的简单字符串解析器

  • 可以将AIMessage解析为简单的字符串，符合了模型invoke方法要求(可传入字符串，不接收AIMessage类型)
  • 是Runnable接口的子类(可以加入链）

from langchain_core.prompts import ChatPromptTemplate,MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_ollama import OllamaLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import AIMessage

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位诗人，可以作诗"),
        MessagesPlaceholder("history"),
        ("human", "非常棒，再来一首诗")
    ]
)

history_data = [
    ("human", "来一首诗"),
    ("ai", "床前明月光，疑似地上霜，举头望明月，低头思故乡"),
    ("human", "好诗，再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦")
]

# model = OllamaLLM(model="qwen3:0.6b")
model = ChatTongyi(model="qwen3-max", api_key="xxx")
parser = StrOutputParser()
# 组成链
# model输出是AIMessage, 将AIMessage传入StrOutputParser，解析成str,再传给model
chain = chat_prompt_template | model | parser | model
res: AIMessage = chain.invoke({"history": history_data})
print(res.content)

JsonOutputParser

model输出的类型是AIMessage类型，所以需要 AIMessage -> dict -> 注入提示词模版形参新的提示词（PromptValue类型）-> model

在构建链的时候要注意整体兼容性，注意前后组件的输入和输出要求。

  • 模型输入:PromptValue或字符串或序列(BaseMessage、list、tuple、str、dict)。
  • 模型输出:AIMessage
  • 提示词模板输入:要求是字典
  • 提示词模板输出:PromptValue对象
  • StrOutputParser:AIMessage输入、str输出
  • Json0utputParser:AIMessage输入、dict输出
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi

str_parser = StrOutputParser()
json_parser = JsonOutputParser

first_prompt = PromptTemplate.from_template(
    "我的邻居性{lastname}, 刚生了{gender}, 帮忙起个名字，以json的格式返回给我"
    "要求key是name，value是名字"
)

second_prompt = PromptTemplate.from_template(
    "姓名{name}，帮我解析的含义"
)

model = ChatTongyi(model="qwen3-max", api_key="xxx")

chain = first_prompt | model | json_parser | second_prompt | model | str_parser
res = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res)

Runnablelambda

RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain.

  • 将函数封装入RunnableLambda类对象，其是Runnable接口实例，可以直接入链
  • 直接将函数入链，函数会自动转换为RunnableLambda对象
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_community.chat_models.tongyi import ChatTongyi

str_parser = StrOutputParser()


first_prompt = PromptTemplate.from_template(
    "我的邻居性{lastname}, 刚生了{gender}, 帮忙起个名字，仅输出名字即可"
)

second_prompt = PromptTemplate.from_template(
    "姓名{name}，帮我解析的含义"
)

model = ChatTongyi(model="qwen3-max", api_key="xxx")
# 方式 1
# 这里的ai_msg是chain链中前一个model输出的AIMessage
my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})
chain = first_prompt | model | my_func | second_prompt | model | str_parser
# 方式 2
chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser
res = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res)

Memory

如果想要封装历史记录，除了自行维护历史消息外，也可以借助LangChain内置的历史记录附加功能。LangChain提供了History功能，帮助模型在有历史记忆的情况下回答。

  • RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链(新Runnable实例)
  • InMemoryChatMessaqeHistory为历史会话记录提供内存存储(临时用),langchain自带
  • FileChatMessageHistory为历史话记录提供文件存储(长期记录)，需自行实现

 InMemoryChatMessaqeHistory 临时会话记忆

    常见使用方法1：

# 创建一个消息历史实例
history = InMemoryChatMessageHistory()

# 添加消息
history.add_message(HumanMessage(content="你好！"))
history.add_message(AIMessage(content="你好呀！有什么我可以帮你的吗？"))

# 查看所有消息
print(history.messages)
# 输出：
# [
#   HumanMessage(content='你好！'),
#   AIMessage(content='你好呀！有什么我可以帮你的吗？')
# ]

# 清空历史
history.clear()

#-----------------------------------------

常见使用方法2：
# 获取指定回话ID的历史会话函数
chat_history_store = {}  # 存放多个会话ID所对应的历史会话记录

def get_history(session_id: str) -> InMemoryChatMessageHistory:
    """
    根据会话ID返回BaseChatMessageHistory的子类：InMemoryChatMessageHistory

    BaseChatMessageHistory类专用于存储会话的历史记录
    InMemoryChatMessageHistory类是官方自带的基于内存存储历史记录的类

    :param session_id: 会话ID
    :return: InMemoryChatMessageHistory
    """
    if session_id not in chat_history_store:
        chat_history_store[session_id]= InMemoryChatMessageHistory()
    return chat_history_store[session_id]

# 通过 RunnableWithMessageHistory 获取一个新的带有历史记录功能的chain
conversation_chain = RunnableWithMessageHistory(
    some_chain,    # 被附加历史消息的 Runnable， 通常是chain
    get_history,  # 获取指定回话ID的历史会话函数
    input_messages_key="input",  # 声明用户消息在模版中的占位符
    history_messages_key="chat_history"  # 声明历史消息在模版中的占位符
)

    案例：

    
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser


from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory


model = ChatTongyi(model="qwen3-max", api_key="sk-xxx")
# prompt = PromptTemplate.from_template(
#     "你需要根据历史会话回应用户问题，对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据历史会话回应用户问题，对话历史:"),
        MessagesPlaceholder("chat_history"),
        ("human", "用户提问：{input}，请回答")
    ]
)
str_parser = StrOutputParser()

def debug_print_prompt(full_prompt):
    print("="*10, full_prompt.to_string(), "="*10)
    return full_prompt

base_chain = prompt | debug_print_prompt | model | str_parser


# 创建有一个新的链，对原有链增强功能：自动附加历史消息

# 获取指定回话ID的历史会话函数
chat_history_store = {}  # 存放多个会话ID所对应的历史会话记录, key是session_id, value是InMemoryChatMessageHistory类对象

def get_history(session_id: str) -> InMemoryChatMessageHistory:
    """
    根据会话ID返回BaseChatMessageHistory的子类：InMemoryChatMessageHistory

    BaseChatMessageHistory类专用于存储会话的历史记录
    InMemoryChatMessageHistory类是官方自带的基于内存存储历史记录的类

    :param session_id: 会话ID
    :return: InMemoryChatMessageHistory
    """
    if session_id not in chat_history_store:
        chat_history_store[session_id]= InMemoryChatMessageHistory()
    return chat_history_store[session_id]

conversation_chain= RunnableWithMessageHistory(
    base_chain,  # 被增强的链
    get_history, # 根据会话ID返回InMemoryChatMessageHistory类的函数
    input_messages_key="input",  # 用户输入在模版中的占位符
    history_messages_key="chat_history"  # 历史消息在模版中的占位符
)

if __name__ == '__main__':
    # 固定格式，添加langchain的配置，为当前程序配置所属的session_id
    # 添加此此配置后，RunnableWithMessageHistory再调用get_history方法时才会传入当前程序配置的session_id，否则报错
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }
    res = conversation_chain.invoke({"input": "小明有2只猫"}, session_config)
    print("第一次执行结果：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第二次执行结果：", res)

    res = conversation_chain.invoke({"input": "共有几只动物?"}, session_config)
    print("第三次执行结果：", res)

    FileChatMessaqeHistory 长期会话记忆

    FileChatMessageHistory类实现，核心思路

    • 以session_id为文件名，不同session_id有不同文件存储消息
    • 继承BaseChatMessageHistory实现如下3个方法:
    1. add messages:同步模式，添加消息
    2. messages:同步模式，获取消息
    3. clear:同步模式，清除消息
    import os.path
from typing import Sequence

from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory

'''
message_to_dict 将单个消息BaseMessage对象 换成 dict
messages_from_dict [dict,dict ...]  ->  [BaseMessage，BaseMessage ...]
'''

class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id  # 会话ID
        self.storage_path = storage_path  # 会话文件存储位置的文件夹路径

        # 完整的会话文件路径
        self.file_path = os.path.join(storage_path, session_id)

        # 确保文件夹存在
        os.makedirs(storage_path, exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        '''
        将传入的messages追加到已有的messages列表中
        :param messages: 传入的messages序列
        :return:
        '''
        # self.messages是父类BaseChatMessageHistory中的元素，表示已有的消息列表 [BaseMessage,BaseMessage ...]
        all_messages = list(self.messages)
        all_messages.extend(messages)  # 追加消息

        # 将数据同步写入到本地文件中
        new_messages = []
        for message in all_messages:
            new_messages.append(message_to_dict(message))  # [dict,dict ...]
        # 将数据写入文件
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump(new_messages, f)

    @property
    def messages(self) -> list[BaseMessage]:
        '''
        打开文件，读取内容，返回 list[BaseMessage]
        :return:
        '''
        try:
            with open(self.file_path, 'r', encoding='utr-8') as f:
                 message_data = json.load(f)
                 return messages_from_dict(message_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump([], f)

    案例

    import json
import os.path
from typing import Sequence

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory


class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):

        self.session_id = session_id  # 会话ID
        self.storage_path = storage_path  # 会话文件存储位置的文件夹路径

        # 完整的会话文件路径
        self.file_path = os.path.join(storage_path, session_id)

        # 确保文件夹存在
        os.makedirs(storage_path, exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        '''
        将传入的messages追加到已有的messages列表中
        :param messages: 传入的messages序列
        :return:
        '''
        # self.messages是父类BaseChatMessageHistory中的元素，表示已有的消息列表 [BaseMessage,BaseMessage ...]
        all_messages = list(self.messages)
        all_messages.extend(messages)  # 追加消息

        # 将数据同步写入到本地文件中
        new_messages = []
        for message in all_messages:
            new_messages.append(message_to_dict(message))  # [dict,dict ...]
        # 将数据写入文件
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump(new_messages, f)

    @property
    def messages(self) -> list[BaseMessage]:
        '''
        打开文件，读取内容，返回 list[BaseMessage]
        :return:
        '''
        try:
            with open(self.file_path, 'r', encoding='utf-8') as f:
                 message_data = json.load(f)
                 return messages_from_dict(message_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump([], f)





model = ChatTongyi(model="qwen3-max", api_key="sk-xxx")
# prompt = PromptTemplate.from_template(
#     "你需要根据历史会话回应用户问题，对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据历史会话回应用户问题，对话历史:"),
        MessagesPlaceholder("chat_history"),
        ("human", "用户提问：{input}，请回答")
    ]
)
str_parser = StrOutputParser()

def debug_print_prompt(full_prompt):
    print("="*10, full_prompt.to_string(), "="*10)
    return full_prompt

base_chain = prompt | debug_print_prompt | model | str_parser


# 创建有一个新的链，对原有链增强功能：自动附加历史消息

def get_history(session_id: str) -> FileChatMessageHistory:
    return FileChatMessageHistory(session_id, './chat_history')

conversation_chain= RunnableWithMessageHistory(
    base_chain,  # 被增强的链
    get_history, # 根据会话ID返回FileChatMessageHistory类的函数
    input_messages_key="input",  # 用户输入在模版中的占位符
    history_messages_key="chat_history"  # 历史消息在模版中的占位符
)

if __name__ == '__main__':
    # 固定格式，添加langchain的配置，为当前程序配置所属的session_id
    # 添加此此配置后，RunnableWithMessageHistory再调用get_history方法时才会传入当前程序配置的session_id，否则报错
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }
    res = conversation_chain.invoke({"input": "小明有2只猫"}, session_config)
    print("第一次执行结果：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第二次执行结果：", res)
    #
    res = conversation_chain.invoke({"input": "共有几只动物?"}, session_config)
    print("第三次执行结果：", res)

    ConversationBufferWindowMemory

    ConversationBufferWindowMemory 是 LangChain 中用于实现多轮会话记忆的一个轻量级 Memory 模块。
    它的核心思想是：只保留最近的 N 轮对话历史（即“滑动窗口”），避免随着对话变长而无限增长上下文长度。

    案例：

    # from langchain.memory import ConversationBufferWindowMemory
# from langchain.chains import ConversationChain

from langchain_classic.memory.buffer_window import ConversationBufferWindowMemory
from langchain_classic.chains import ConversationChain

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_ollama import OllamaLLM
from langchain_core.prompts import PromptTemplate

'''
ConversationBufferWindowMemory 是 LangChain 中用于实现多轮会话记忆的一个轻量级 Memory 模块。
它的核心思想是：只保留最近的 N 轮对话历史（即“滑动窗口”），避免随着对话变长而无限增长上下文长度。
| 参数 | 类型 | 说明 |
|------|------|------|
| `k` | int | 保留最近多少轮对话（每轮包含 human + ai 一对） |
| `return_messages` | bool | 是否以 `BaseMessage` 对象形式返回（默认 False，返回字符串） |
| `memory_key` | str | 在 prompt 中引用的 key，默认为 `"history"` |
| `input_key` / `output_key` | str | 指定输入/输出字段名（用于复杂 chain） |

'''
# 创建一个只记住最近 2 轮对话的 memory
memory = ConversationBufferWindowMemory(k=3)

# 初始化 LLM
# model = ChatTongyi(model="qwen3-max", api_key="sk-xxx")
model = OllamaLLM(model="deepseek-r1:1.5b", base_url="http://172.30.18.102:11434",)

# 创建带 memory 的对话链
conversation = ConversationChain(
    llm=model,
    memory=memory,
    # prompt=prompt,
    verbose=True  # 可选：打印内部 prompt 查看上下文
)


response1 = conversation.predict(input="你好！")
print(response1)

response2 = conversation.predict(input="我叫小明。")
print(response2)

response3 = conversation.predict(input="我喜欢篮球。")
print(response3)

response4 = conversation.predict(input="你还记得我叫什么吗？")
print(response4)

# 查看当前记忆内容 输出示例：{'history': 'Human: 你好！\nAI: 你好呀！\nHuman: 我叫小明。\nAI: 很高兴认识你，小明！'}
print(memory.load_memory_variables({}))
    特性 InMemoryChatMessageHistory ConversationBufferWindowMemory
    类型 底层消息存储（List[BaseMessage]） 高层 Memory 模块（字符串格式）
    窗口限制 ❌ 无（除非手动截断） ✅ 支持 k 轮滑动窗口
    与 Chain 集成 需配合 RunnableWithMessageHistory 直接用于 ConversationChain
    多会话支持 ✅ 通过 session_id 管理 ❌ 通常单会话
    消息格式 BaseMessage 对象（结构化） 字符串（"Human: ...\nAI: ..."）

    trim_messages

    trim_messages 是 LangChain 提供的一个用于自动裁剪对话历史以控制 token 长度的实用工具，特别适用于使用 InMemoryChatMessageHistory 时防止上下文过长导致 LLM 超出最大 token 限制（如 Qwen、GPT 等模型通常有 32768 或更少的上下文窗口）

    场景 说明
    长对话累积 用户连续聊天几十轮，历史消息超出模型上下文长度
    避免报错 防止 400 Bad Request: prompt too long 错误
    成本控制 减少不必要的 token 消耗，节省 API 费用
    实时流式对话 在每次用户输入前动态裁剪历史 
    from langchain_core.messages.utils import trim_messages
from langchain_openai import ChatOpenAI


trimmed = trim_messages(
    messages,                # List[BaseMessage]：原始消息列表
    max_tokens=3000,         # 最大允许 token 数
    strategy="last",         # 裁剪策略："last"（保留最新）、"first"（保留最早）
    token_counter=ChatOpenAI(model="gpt-4o")，
    include_system=True,     # 是否包含 system 消息（默认 True）
)

    案例：

    from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages.utils import trim_messages
from langchain_core.runnables import RunnableLambda
from langchain_core.messages import BaseMessage

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from transformers import AutoTokenizer


model = ChatTongyi(model="qwen3-max", api_key="sk-xxx")

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据历史会话回应用户问题，对话历史:"),
        MessagesPlaceholder("chat_history"),
        ("human", "用户提问：{input}，请回答")
    ]
)
str_parser = StrOutputParser()

# 1. 从本地路径加载模型和 tokenizer
local_model_path = "./qwen-7b-tokenizer/"
tokenizer = AutoTokenizer.from_pretrained(
    local_model_path,
    trust_remote_code=True,
    local_files_only=True  # 强制离线
)
def count_tokens(msg):
    return tokenizer(msg.content, return_length=True)['length'][0]

# ===== 新增：裁剪函数 =====
def trim_history(messages: list[BaseMessage]) -> list[BaseMessage]:
    """
    对 prompt 生成的完整消息列表进行裁剪，保留最近内容，不超过 token 限制
    """
    return trim_messages(
        messages,
        max_tokens=6000,          # 留出空间给模型输出（Qwen 最大 32768，这里保守）
        strategy="last",          # 保留最新消息
        token_counter=lambda msg: len(tokenizer.encode(msg.content)),  # 直接使用 lambda
        include_system=True,      # 系统消息始终保留
    )

def debug_print_prompt(full_prompt):
    print("="*10, full_prompt.to_string(), "="*10)
    return full_prompt

# 构建 base_chain：prompt → 裁剪 → model → str
base_chain = prompt | RunnableLambda(trim_history) | debug_print_prompt | model | str_parser


# 创建有一个新的链，对原有链增强功能：自动附加历史消息

# 获取指定回话ID的历史会话函数
chat_history_store = {}  # 存放多个会话ID所对应的历史会话记录, key是session_id, value是InMemoryChatMessageHistory类对象

def get_history(session_id: str) -> InMemoryChatMessageHistory:
    """
    根据会话ID返回BaseChatMessageHistory的子类：InMemoryChatMessageHistory

    BaseChatMessageHistory类专用于存储会话的历史记录
    InMemoryChatMessageHistory类是官方自带的基于内存存储历史记录的类

    :param session_id: 会话ID
    :return: InMemoryChatMessageHistory
    """
    if session_id not in chat_history_store:
        chat_history_store[session_id]= InMemoryChatMessageHistory()
    return chat_history_store[session_id]

conversation_chain= RunnableWithMessageHistory(
    base_chain,  # 被增强的链
    get_history, # 根据会话ID返回InMemoryChatMessageHistory类的函数
    input_messages_key="input",  # 用户输入在模版中的占位符
    history_messages_key="chat_history"  # 历史消息在模版中的占位符
)

if __name__ == '__main__':
    # 固定格式，添加langchain的配置，为当前程序配置所属的session_id
    # 添加此此配置后，RunnableWithMessageHistory再调用get_history方法时才会传入当前程序配置的session_id，否则报错
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }
    res = conversation_chain.invoke({"input": "小明有2只猫"}, session_config)
    print("第一次执行结果：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第二次执行结果：", res)

    res = conversation_chain.invoke({"input": "共有几只动物?"}, session_config)
    print("第三次执行结果：", res)

"""
工作流程说明：
用户输入 {"input": "..."}
RunnableWithMessageHistory 自动从 get_history("user_001") 拿到历史消息（List[BaseMessage]）
注入到 prompt 中的 MessagesPlaceholder("chat_history")
prompt 输出完整的 List[BaseMessage]（包含 system + history + current human）
trim_history 对该列表裁剪，确保总 token ≤ 6000
裁剪后的消息送入 ChatTongyi
结果解析为字符串返回
✅ 所有历史仍完整保存在 InMemoryChatMessageHistory 中（用于未来裁剪参考），但每次推理只用裁剪后的子集。
"""

    下载tokenizer到本地使用

    from transformers import GPT2Tokenizer

# 指定要下载的模型名称
model_name = "Qwen/Qwen-7B-Cha"

# 1. 从HuggingFace加载tokenizer（此时会从网络下载）
tokenizer = GPT2Tokenizer.from_pretrained(model_name)

# 2. 将tokenizer的文件保存到你指定的本地目录
# 请将 './gpt2_tokenizer/' 替换为你想要保存的路径
save_directory = "./qwen-7b-tokenizer/"
tokenizer.save_pretrained(save_directory)

print(f"Tokenizer 文件已成功下载并保存到: {save_directory}")

    生产环境一般使用自定义的裁剪策略，实现对历史消息的 存取、摘要、压缩等操作，如下：

    from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages.utils import trim_messages
from langchain_core.runnables import RunnableLambda
from langchain_core.messages import BaseMessage, get_buffer_string

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from transformers import AutoTokenizer


model = ChatTongyi(model="qwen3-max", api_key="sk-xxx")

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据历史会话回应用户问题，对话历史:"),
        MessagesPlaceholder("chat_history"),
        ("human", "用户提问：{input}，请回答")
    ]
)
str_parser = StrOutputParser()

# 1. 从本地路径加载模型和 tokenizer
local_model_path = "./qwen-7b-tokenizer/"
tokenizer = AutoTokenizer.from_pretrained(
    local_model_path,
    trust_remote_code=True,
    local_files_only=True  # 强制离线
)
def count_tokens(msg):
    return tokenizer(msg.content, return_length=True)['length'][0]

# ===== 新增：裁剪函数 =====
# def trim_history(messages: list[BaseMessage]) -> list[BaseMessage]:
#     """
#     对 prompt 生成的完整消息列表进行裁剪，保留最近内容，不超过 token 限制
#     """
#     return trim_messages(
#         messages,
#         max_tokens=6000,          # 留出空间给模型输出（Qwen 最大 32768，这里保守）
#         strategy="last",          # 保留最新消息
#         token_counter=lambda msg: len(tokenizer.encode(msg.content)),  # 直接使用 lambda
#         include_system=True,      # 系统消息始终保留
#     )
def trim_history(messages: list[BaseMessage]) -> list[BaseMessage]:
    full_text = get_buffer_string(messages)
    if count_tokens(full_text) <= 6000:
        return messages

    # 分离 system 消息
    system_msgs = [m for m in messages if m.type == "system"]
    other_msgs = [m for m in messages if m.type != "system"]

    kept = system_msgs.copy()
    current_tokens = sum(count_tokens(m.content) for m in kept)

    # 从最新消息开始往前加（保留最近对话）
    for msg in reversed(other_msgs):
        msg_tokens = count_tokens(msg.content)
        if current_tokens + msg_tokens > 6000:
            break
        kept.insert(len(system_msgs), msg)  # 插入到 system 之后
        current_tokens += msg_tokens

    return kept

def debug_print_prompt(full_prompt):
    print("="*10, full_prompt.to_string(), "="*10)
    return full_prompt

# 构建 base_chain：prompt → 裁剪 → model → str
base_chain = prompt | RunnableLambda(trim_history) | debug_print_prompt | model | str_parser


# 创建有一个新的链，对原有链增强功能：自动附加历史消息

# 获取指定回话ID的历史会话函数
chat_history_store = {}  # 存放多个会话ID所对应的历史会话记录, key是session_id, value是InMemoryChatMessageHistory类对象

def get_history(session_id: str) -> InMemoryChatMessageHistory:
    """
    根据会话ID返回BaseChatMessageHistory的子类：InMemoryChatMessageHistory

    BaseChatMessageHistory类专用于存储会话的历史记录
    InMemoryChatMessageHistory类是官方自带的基于内存存储历史记录的类

    :param session_id: 会话ID
    :return: InMemoryChatMessageHistory
    """
    if session_id not in chat_history_store:
        chat_history_store[session_id]= InMemoryChatMessageHistory()
    return chat_history_store[session_id]

conversation_chain= RunnableWithMessageHistory(
    base_chain,  # 被增强的链
    get_history, # 根据会话ID返回InMemoryChatMessageHistory类的函数
    input_messages_key="input",  # 用户输入在模版中的占位符
    history_messages_key="chat_history"  # 历史消息在模版中的占位符
)

if __name__ == '__main__':
    # 固定格式，添加langchain的配置，为当前程序配置所属的session_id
    # 添加此此配置后，RunnableWithMessageHistory再调用get_history方法时才会传入当前程序配置的session_id，否则报错
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }
    res = conversation_chain.invoke({"input": "小明有2只猫"}, session_config)
    print("第一次执行结果：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第二次执行结果：", res)

    res = conversation_chain.invoke({"input": "共有几只动物?"}, session_config)
    print("第三次执行结果：", res)

    文档加载器 

    文档加载器提供了一套标准接口，用于将不同来源(如CSV、PDF或JSON等)的数据读取为 Langchain的文档格式。这确保了无论数据来源如何，都能对其进行一致性处理。文档加载器(内置或自行实现)需实现BaseLoader接口，

    不同的文档加载器可能定义了不同的参数，但是其都实现了统一的接口(方法)·

    • load():一次性加载全部文档
    • lazy_load():延迟流式传输文档，对大型数据集很有用，避免内存溢出。

    CSVLoader

    将csv文本数据加载为Document类型对象

    csvfile.csv:
name,age,hobby
小明,20,"唱，跳"
小刚,22,"rap"


from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path='./docs/csvfile.csv',  # csv文件路径
    encoding='utf-8',  # 指定编码
    csv_args={
        "delimiter": ',',  # 指定分隔符
        "quotechar": '"',  #
        # "fieldnames": ['name','age','爱好']  # 若元数据没有表头则使用此参数
    }
)

# 批量加载 .load() -> [Document,Document ...]
documents = loader.load()
print(documents)

# 懒加载 .lazy_load() -> 迭代器
documents = loader.lazy_load()
for doc in documents:
    print(doc)

    JSONLoader

    JSONLoader用于将JSON数据加载为Document类型对象。

    使用JSONLoader需要额外安装:pip install jq

    jq是一个跨平台的json解析工具，LangChain底层对JSON的解析就是基于iq工具实现的。将JSON数据的信息抽取出来，封装为Document对象，抽取的时候依赖jqschema语法

    stu.json:
{
  "name":"鸡哥",
  "hobby": ["唱","跳","rap"],
  "addr": {
    "c": "上海"
  }
}


from langchain_community.document_loaders import JSONLoader


loader = JSONLoader(
    file_path="./docs/stu.json",
    jq_schema=".",  # jq解析语法
    text_content=False,  # 当获取的值不是字符串时需要指定为False
    json_lines=True  # 当文件内容是 jsonLines 格式时需要指定为True， jsonLines：每行是一个dict
)

documents = loader.load()
print(documents)

documents = loader.lazy_load()
for doc in documents:
    print(doc)

    TextLoader

    作用:读取文本文件(如.txt)，将全部内容放入一个Document对象中,注意：返回仅有一个Document对象的list

    from langchain_community.document_loaders import TextLoader

loader = TextLoader(
    file_path="./docs/python.txt",
    encoding="utf-8"
)

documents = loader.load()  # TextLoader将文本内容读到一个Document对象中
print(documents)  # [Document]

    PyPDFLoader

    from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(
    file_path="./docs/python.pdf",
    mode="page",  # 默认 page 模式，每一页是一个Documents，single模式整个pdf是一个Documents
    password="123456", # 若pdf有密码可使用此参数传入密码
)

documents = loader.lazy_load()
for document in documents:
    print(document)

    RecursiveCharacterTextSplitter

    from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
loader = TextLoader(
    file_path="./docs/python.txt",
    encoding="utf-8"
)

documents = loader.load()
print(documents)

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # 分段的最大字符数
    chunk_overlap=50,  # 分段之间允许的重叠字符数
    separators=["\n\n", "\n", ".", ",", "?", "。", "，", "？"],  # 文本段落分割的符号依据
    length_function=len  # 统计字符长度的依据函数
)

split_docs = splitter.split_documents(documents)

    vector stores向量存储

    基于LangChain的向量存储，存储嵌入数据，并执行相似性搜索，如下图是典型的RAG流程：

    LangChain为向量存储提供了统一接口:

    • add documents
    • delete
    • similarity search

    InMemoryVectorStore

    内存向量存储

    from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings

# 初始化向量存储
vector_stores = InMemoryVectorStore(
    embedding=DashScopeEmbeddings()  # 使用的文本转向量模型
)
# 添加文档到向量存储并指定ID
vector_stores.add_documents(documents=[doc1, doc2], ids=[1,2])  # 为每个doc创建一个唯一ID
# 删除文档（根据ID删除）
vector_stores.delete(ids=[1])
# 相似度检索
vector_stores.similarity_search("your query", 5)  # k参数表示返回几条内容

    案例

    import uuid

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path='./docs/csvfile.csv',  # csv文件路径
    encoding='utf-8',  # 指定编码
    csv_args={
        "delimiter": ',',  # 指定分隔符
        "quotechar": '"',  #
        # "fieldnames": ['name','age','爱好']  # 若元数据没有表头则使用此参数
    }
)

# 批量加载 .load() -> [Document,Document ...]
documents = loader.load()
# 为每个Document创建唯一ID,可以使用uuid
ids = [index for index, _ in enumerate(documents)]

# 初始化向量存储
vector_stores = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(),  # 使用的文本转向量模型
)
# 添加文档到向量存储并指定ID
vector_stores.add_documents(documents=documents, ids=ids)  # 为每个doc创建一个唯一ID
# 删除文档（根据ID删除）
vector_stores.delete(ids=[1])
# 相似度检索
query_documents = vector_stores.similarity_search("唱", 3)  # k参数表示返回几条内容

    Chroma

    from langchain_community.embeddings import DashScopeEmbeddings
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma

# pip3.11 install sentence-transformers langchain-huggingface
# 配置本地嵌入模型（会自动下载到本地缓存，或从指定路径加载）
embedding_model = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",  # 中文推荐模型
    # model_kwargs={"device": "cuda"},    # 如果有 GPU 可启用
    # cache_folder="./models"             # 可选：指定模型缓存目录
)

# 初始化向量数据库
vector_stores = Chroma(
    collection_name="collection",  # 聚集，相当于表名
    embedding_function=embedding_model,  # 使用的文本转向量模型
    # embedding_function=DashScopeEmbeddings(),  # 使用的文本转向量模型
    persist_directory="./chroma_db"  # 存储位置
)

    案例

    from langchain_community.embeddings import DashScopeEmbeddings
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path='./docs/csvfile.csv',  # csv文件路径
    encoding='utf-8',  # 指定编码
    csv_args={
        "delimiter": ',',  # 指定分隔符
        "quotechar": '"',  #
        # "fieldnames": ['name','age','爱好']  # 若元数据没有表头则使用此参数
    }
)

# 批量加载 .load() -> [Document,Document ...]
documents = loader.load()
# 为每个Document创建唯一ID,可以使用uuid
ids = [index for index, _ in enumerate(documents)]

# 配置本地嵌入模型（会自动下载到本地缓存，或从指定路径加载）
embedding_model = HuggingFaceEmbeddings(
    model_name="bge-small-zh-v1.5",  # 中文推荐模型
    # model_kwargs={"device": "cuda"},    # 如果有 GPU 可启用
    # cache_folder="./models"             # 可选：指定模型缓存目录
)

# 初始化向量数据库
vector_stores = Chroma(
    collection_name="collection",  # 聚集，相当于表名
    embedding_function=embedding_model,  # 使用的文本转向量模型
    # embedding_function=DashScopeEmbeddings(),  # 使用的文本转向量模型
    persist_directory="./chroma_db"  # 存储位置
)
# 添加文档到向量存储并指定ID
vector_stores.add_documents(documents=documents, ids=ids)  # 为每个doc创建一个唯一ID
# 删除文档（根据ID删除）
vector_stores.delete(ids=[1])
# 相似度检索
query_documents = vector_stores.similarity_search(
    "唱",
    3,  # k参数表示返回几条内容
    filter={"name": "小明"}  # 过滤
)
print(query_documents)

    RunnablePassthrough

    from typing import List

from langchain_community.chat_models import ChatTongyi
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document

api_key="sk-xxx"

# 模型
model = ChatTongyi(model="qwen3-max", api_key=api_key)
# 提示词
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "根据提供的产开资料为主，简洁专业的回答用户问题，参考资料：{context}"),
        ("user", "用户问题：{input}")
    ]
)

# 向量数据库
embedding_model = HuggingFaceEmbeddings(
    model_name="bge-small-zh-v1.5",  # 中文推荐模型
    # model_kwargs={"device": "cuda"},    # 如果有 GPU 可启用
    # cache_folder="./models"             # 可选：指定模型缓存目录
)
# 这里使用内存向量数据库
vector_store = InMemoryVectorStore(
    # embedding=DashScopeEmbeddings(model="text-embedding-v4")
    embedding=embedding_model
)
# 添加文档到向量数据库
vector_store.add_texts(["减肥就是需要少吃多练", "在减肥期间需要控油腻食物，减少卡路里的摄入，并且按时运行", "跑步时很好的有氧运动"])

input_text = "如何减肥"

# 从向量数据库中获取参考资料
res = vector_store.similarity_search(input_text, 2)

# 组织参考资料
reference_text = "["
for doc in res:
    reference_text += doc.page_content
reference_text += "]"

# chain
def debug_print_prompt(prompt):
    print("="*10)
    print(prompt.to_string())
    print("="*10)
    return prompt

chain = prompt | debug_print_prompt | model | StrOutputParser()
result = chain.stream(input={"input": input_text, "context": reference_text})
for output in result:
    print(output, end="", flush=True)
# result = chain.invoke({"input": input_text, "context": reference_text})
# print(result)

'''
将向量检索入链
1. 如链的前提是需要Runnable的子类
2. 链组件的输入输出：
    向量检索：
        输入：用户的提问                str
        输出：向量检索结果              List[Document]
    prompt:
        输入：用户提问+向量检索结果      dict
        输出：完整的提示词              PromptValue
'''


# 向量对象转成Runnable对象
retriever = vector_store.as_retriever(searech_kwargs={"k": 2})
def format_func(docs: List[Document]):
    if not docs:
        return "无参考资料"
    reference_text = "["
    for doc in res:
        reference_text += doc.page_content
    reference_text += "]"
    return reference_text
chain = (
    {"input": RunnablePassthrough(), "context": retriever | format_func} | prompt | debug_print_prompt | model | StrOutputParser()
)

chain.invoke(input=input_text)

    RAG实战

    RAG即检索、增强和生成，其主要分为2条线:

    • 离线处理:向私有知识库(向量存储)源源不断添加私有知识文档。向知识库添加来自未来的知识文档(基于模型训练完成时间)向模型添加私有知识文档,给出模型参考资料，规避模型幻觉(一本正经的胡说八道，
    • 在线处理:用户提问会先基于私有知识库做检索，获取参考资料，同步组装新提示词询问大模型获取结果。

    Logo
    CSDN-OPC开发者社区

    这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。

    更多推荐

    cover

    从 “只会聊天” 到 “动手干活”：全面科普 OpenClaw 小龙虾 AI 智能体

    avatar CSDN-OPC开发者社区
    cover

    【小白也能轻松玩转龙虾】虾壳云一键部署全套攻略，完整走完 OpenClaw v2.7.9 安装全流程（附最新安装包）

    avatar CSDN-OPC开发者社区
    cover

    开源本地 AI 智能体网关 OpenClaw 深度实践：架构解析、全场景部署与自动化落地指南

    avatar CSDN-OPC开发者社区