from typing import Callable, Optional import gradio as gr from langchain.embeddings.openai import OpenAIEmbeddings from langchain.vectorstores import Zilliz from langchain.document_loaders import TextLoader from langchain.text_splitter import CharacterTextSplitter from langchain.chains import RetrievalQAWithSourcesChain import uuid from project.llm.zhipuai_llm import ZhipuAILLM from project.prompt.answer_by_private_prompt import ( COMBINE_PROMPT, EXAMPLE_PROMPT, QUESTION_PROMPT, ) chain: Optional[Callable] = None def web_loader(file, openai_key, puzhiai_key, zilliz_uri, user, password): if not file: return "please upload file" loader = TextLoader(file) docs = loader.load() text_splitter = CharacterTextSplitter(chunk_size=1024, chunk_overlap=0) docs = text_splitter.split_documents(docs) embeddings = OpenAIEmbeddings(model="text-embedding-ada-002", openai_api_key=openai_key) if not embeddings: return "embeddings not" texts = [d.page_content for d in docs] docsearch = Zilliz.from_documents( docs, embedding=embeddings, ids=[str(uuid.uuid4()) for _ in range(len(texts))], connection_args={ "uri": zilliz_uri, "user": user, "password": password, "secure": True, }, ) if not docsearch: return "docsearch not" global chain #chain = RetrievalQAWithSourcesChain.from_chain_type( # ZhipuAILLM(model="glm-3-turbo", temperature=0.1, zhipuai_api_key=puzhiai_key), # chain_type="refine", # retriever=docsearch.as_retriever(), #) chain = RetrievalQAWithSourcesChain.from_llm( ZhipuAILLM(model="glm-3-turbo", temperature=0.1, zhipuai_api_key=puzhiai_key), EXAMPLE_PROMPT, QUESTION_PROMPT, COMBINE_PROMPT, retriever=docsearch.as_retriever(), ) return "success to load data" def query(question): global chain # "What is milvus?" if not chain: return "please load the data first" return chain(inputs={"question": question}, return_only_outputs=True).get( "answer", "fail to get answer" ) if __name__ == "__main__": block = gr.Blocks() with block as demo: gr.Markdown( """