GPTS Builder

GPTS Builder 是一个高度模块化的 Python 库，专为快速构建和部署基于大型语言模型（如 GPT-3 或 GPT-4）的应用而设计。它提供了一系列工具，使开发者能够轻松管理会话、创建提示词模板、构建对话流程，并进行配置管理。

特点

• 上下文管理器：有效管理对话状态，支持多种存储模式。
• 提示词模板构建器：快速生成和管理提示词模板，提高对话生成的效率和质量。
• 大模型对话构建器：继承自提示词模板构建器，提供高效的对话管理。
• 配置管理器：灵活配置模型参数，包括模型名称和最大 Token 数量。
• 存储模式支持：支持全局变量和 Redis 存储，确保灵活性和扩展性。

架构概览

GPTS Builder 采用模块化设计，主要包括以下四个核心组件：

1. 上下文管理器：管理用户会话和状态，支持全局变量和 Redis 存储模式。
2. 提示词模板构建器：构建和维护用于生成模型提示词的模板。
3. 大模型对话构建器：基于提示词模板生成对话，支持快速部署和测试模型应用。
4. 配置管理器：中心化管理所有模型和应用配置。

开发计划

知识库集成：将提供与大型知识库的集成，增强模型的应答能力。 GPTS缓存：为了提高响应速度和减少API调用成本，将实现请求结果的缓存机制。

快速开始

LLM 部分

大模型对话构建器

大模型对话构建器继承自提示词模板构建器，集成模板管理与对话生成的功能，支持复杂的对话流程控制，使得生成的对话更加自然和流畅。

可用类：

• LLM: 大模型对话管理。
• LLMAsync: 异步大模型对话管理。

from gpts_builder.builder import LLM
from gpts_builder.builder_async import LLMAsync
from gpts_builder.session_manager.storage.redis_storage_async import RedisStorageAsync

from gpts_builder.config import config_manager

from gpts_builder.util.logger import logger

import asyncio

"""
大语言模型的demo
"""

‘’‘python
# 查看所有模型的配置
configs = config_manager.list_models_config()
print(configs)
config_manager.base_url = "https://www.lazygpt.cn/api【可以改成自己的url，我这里使用的代理】"
config_manager.apikey = "lazygpt-XXXXXXXXXXXX"【改成自己的apikey】


async def llm_async_demo():
    """
    llm_async_demo 测试异步LLM的demo
    """
    ### example0: 初始化LLM（开启一轮对话）
    # 异步的session存储需要设置，目前只支持异步redis存储
    session_storage = RedisStorageAsync("redis://localhost:6379")
    llm = LLMAsync(model="gpt-3.5-turbo", session_storage=session_storage)
    # 设置系统提示词和用户输入
    await llm.set_system("你是一个AI助理").set_prompt("测试回复").build()

    ### example1: 测试一轮对话，返回所有结果
    # 获取非流式回复
    reply = await llm.chat_completions()
    print(f"Received reply {reply}")
    content = reply.get("choices")[0].get("message",{}).get("content") if reply.get("choices") else ""
    print(f"Received content {content}")

    ### example2: 测试一轮对话，流式返回所有结果（获取流式回复：流式返回为一个异步生成器，需要迭代生成）
    async for content in llm.chat_completions_stream():
        print(f"Received content: {content}")
    
    ### example3: 查看当前会话的历史消息
    session = llm.session
    print(f"Session messages: {session.to_json()}")


def llm_sync_demo():
    """
    llm_sync_demo 测试同步LLM的demo
    """
    ### example0: 初始化LLM（开启一轮对话）
    # session存储支持redis存储｜全局变量存储，如果不传session_storage则默认使用全局变量存储
    llm = LLM(model="gpt-3.5-turbo")
    llm.set_system("这里填系统提示词").set_prompt("这里填提示词模板和参数或者内容").build()

    ### example1: 测试一轮对话，返回所有结果
    replay = llm.chat_completions()
    content = replay.get("choices")[0].get("message",{}).get("content") if replay.get("choices") else ""
    print(f"Received content: {content}")

    ### example2: 测试一轮对话，流式返回所有结果（流式返回为一个异步生成器，需要迭代生成）
    for content in llm.chat_completions_stream():
        print(f"Received content: {content}")

    ### example3: 查看当前会话的历史消息
    session = llm.session
    print(f"Session messages: {session.to_json()}")

    ### example4: 清除当前会话(如果不需要上下文管理，存储设置成全局变量存储，然后手动清除会话即可)
    llm.clear_session()
    

if __name__ == "__main__":
    # asyncio.run(llm_async_demo())
    llm_sync_demo()

数据检索增强模块测试代码

为什么要做检索增强（参考文档）

数据检索增强模块测试代码：为什么要做检索增强（https://platform.openai.com/docs/guides/prompt-engineering/strategy-use-external-tools）

可用类：

• DatasetBuilder: 知识库管理（检索增强用）。
• DatasetBuilderAsync 异步知识库管理（检索增强用）。

from gpts_builder.session_manager.storage.redis_storage import RedisStorage
from gpts_builder.session_manager.storage.redis_storage_async import RedisStorageAsync

from gpts_builder.config import config_manager

import asyncio

config_manager.base_url = "https://www.lazygpt.cn/api【可以改成自己的url，我这里使用的代理】"
config_manager.apikey = "lazygpt-XXXXXXXXXXXX"【改成自己的apikey】


def dataset_demo():
    """
    dataset_demo 检索增强数据集的demo
    """
    from gpts_builder.builder import DatasetBuilder
    from gpts_builder.util import PostgresVector
    # 使用知识库，知识库需要一个向量数据库（目前只支持pgvector）
    ### example1: 初始化向量数据库
    db_driver = PostgresVector(dbname="postgres", user="myuser", password="mypassword", host="localhost", port=5432)
    dataset_builder = DatasetBuilder(db_driver=db_driver)
    dataset_id = dataset_builder.create_dataset("测试知识库")

    ### example2: 根据id获取知识库详情
    dataset_details = dataset_builder.get_dataset(filters={dataset_builder.dataset_schema.id: dataset_id})
    print(f"dataset_details1: {dataset_details}")
    
    ### example3: 根据名称获取知识库详情
    dataset_details = dataset_builder.get_dataset(filters={dataset_builder.dataset_schema.name: "测试知识库"})
    print(f"dataset_details2: {dataset_details}")

    ### example4: 向知识库中增加数据【一个答案（内容）对应多个问题（索引）】
    res = dataset_builder.create_datas(dataset_id, "这是一个答案", ["问题1", "问题2"])
    print(f"datas: {res}")

    ### example5: 根据文本在库中查询相似度
    datas_similarity = dataset_builder.query_similarity(text="这是测试数据", dataset_ids=[1, 2, dataset_id])
    print(f"datas_similarity: {datas_similarity}")

    ### example6: 根据文本在库中正则匹配
    datas_regex = dataset_builder.query_regex(regex=".*问题.*", dataset_ids=[dataset_id])
    print(f"datas_regex: {datas_regex}")

async def dataset_asnyc_demo():
    """
    dataset_asnyc_demo 异步操作知识库模块
    """
    from gpts_builder.builder_async import DatasetBuilderAsync
    from gpts_builder.util import PostgresVectorAsync
    # 使用知识库，知识库需要一个向量数据库（目前只支持pgvector）
    db_driver = PostgresVectorAsync(dbname="postgres", user="myuser", password="mypassword", host="127.0.0.1", port=5432)
    dataset_builder = DatasetBuilderAsync(db_driver=db_driver)
    
    ### example1: 创建知识库
    dataset_id = await dataset_builder.create_dataset("测试知识库")

    ### example2: 通过名称获取知识库详情
    dataset_details = await dataset_builder.get_dataset(filters={dataset_builder.dataset_schema.id: dataset_id})

    ### example3: 根据名称获取知识库详情
    dataset_details = await dataset_builder.get_dataset(filters={dataset_builder.dataset_schema.name: "测试知识库"})
    print(f"dataset_details: {dataset_details}")

    ### example4: 向知识库中增加数据【一个答案（内容）对应多个问题（索引）】
    datas = await dataset_builder.create_datas(dataset_id, "这是一个答案", ["问题1", "问题2"])
    print(f"datas: {datas}")

    ### example5: 根据文本在库中查询相似度
    datas_similarity = await dataset_builder.query_similarity(text="这是测试文本", dataset_ids=[dataset_id])
    print(f"datas_similarity: {datas_similarity}")

    ### example6: 根据文本在库中正则匹配
    datas_regex = await dataset_builder.query_regex(regex="*问题*", dataset_ids=[dataset_id])
    print(f"datas_regex: {datas_regex}")


if __name__ == "__main__":
    #dataset_demo()
    asyncio.run(dataset_asnyc_demo())