我正在尝试将本地Python脚本连接到Azure上托管的Redis Enterprise向量数据库，并使用我将用于实现聊天机器人的向量填充它。我可以用redis_client.ping（）建立连接，但问题出现在VectorStore = Redis.from_texts（texts，embeddings，redis_conn=redis_client）行。我也试过Redis.from_documents和Redis.from_texts_return_keys，但我遇到了同样的问题。发生的错误如下：ValueError：找不到redis_url，请添加包含它的环境变量'REDIS_URL'，或将'redis_url'作为命名参数传递，或ConnectionError：服务器已关闭连接。redis vectorstore.这些错误因Redis.from_texts函数的不同参数而异。当我们将脚本连接到本地Redis和Pinecone数据库时，它工作得很好。我的代码如下：

#db:redis/docker, source:pdf, webapp:streamlit

 

import os

import redis

import streamlit as st

from PyPDF2 import PdfReader

from langchain.embeddings.openai import OpenAIEmbeddings

from langchain.text_splitter import CharacterTextSplitter

from langchain.vectorstores.redis import Redis

from langchain.llms import OpenAI

from langchain.chains.question_answering import load_qa_chain

from langchain.vectorstores import Pinecone

from urllib.parse import quote_plus

import pinecone

import spacy

import redis

from apikey import apikey



os.environ["OPENAI_API_KEY"] = apikey

 

# Postavke za povezivanje na Azure Redis

redis_host = '#######'

redis_port = '#######'

redis_password = '#######'


encoding='utf-8'

decode_responses=True

 


redis_client = redis.StrictRedis(

    host=redis_host,

    port=redis_port,

    password=redis_password,

    ssl=True,

    ssl_cert_reqs=None,  # Za neobavezne SSL certifikate

    ssl_ca_certs=None  # Putanja do CA certifikata ako je potrebno

)

 

try:

              
 # Testiranje pinga na Redis server

    result = redis_client.ping()

    if result:

        st.write("Veza s Redis serverom je uspešno uspostavljena!")

    else:

        st.write("Neuspešno testiranje veze s Redis serverom.")

 

except Exception as e:

    st.write("Došlo je do greške pri povezivanju s Redis serverom:", e)

 


with st.sidebar:

 

    st.title('🤗💬 Chat with your Data')

 

   

 

def split_text_into_chunks(text, max_chunk_size):

    nlp = spacy.load("en_core_web_sm")

    # Tokenizacija teksta na rečenice

    doc = nlp(text)

    #doc=text

    sentences = [sent.text for sent in doc.sents]

 

    chunks = []

    current_chunk = ""

    for sentence in sentences:

        if len(current_chunk) + len(sentence) <= max_chunk_size:

            current_chunk += sentence

        else:

            chunks.append(current_chunk.strip())

            current_chunk = sentence

    chunks.append(current_chunk.strip())

   

    return chunks

 

 

 

def main():

 

    st.header("Chat with your own PDF")

    pdf = 'Serbia.pdf'

    

    if pdf is not None:

 

        pdf_reader = PdfReader(pdf)

 

        texts = []

 

        for page in pdf_reader.pages:

 

            text = page.extract_text()

 

            page_chunks = split_text_into_chunks(text,50)

 

            texts.extend(page_chunks)

 

            st.write(texts)

 

        embeddings = OpenAIEmbeddings()

 

        VectorStore = Redis.from_texts(texts, embeddings, redis_conn=redis_client)

        VectorStore.url = None

 

      

 

        query = st.text_input("Ask questions related to your PDF")

 

        st.write(embeddings)

 

        if query:

            results = VectorStore.similarity_search(query=query, k=3)

            llm = OpenAI()

            chain = load_qa_chain(llm=llm, chain_type="stuff")

            response = chain.run(input_documents=results, question=query)

            st.write(response)

 

 

if __name__ == '__main__':

 

    main()