Spaces:
Sleeping
Sleeping
File size: 5,371 Bytes
c104abf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 |
import os
os.environ['TOKENIZERS_PARALLELISM'] = 'true'
from dotenv import load_dotenv
load_dotenv() # load .env api keys
mistral_api_key = os.getenv("MISTRAL_API_KEY")
print("mistral_api_key", mistral_api_key)
import pandas as pd
from langchain.output_parsers import PandasDataFrameOutputParser
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_mistralai import MistralAIEmbeddings
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from typing import Literal
from langchain_core.prompts import PromptTemplate
from langchain_mistralai import ChatMistralAI
from pathlib import Path
from langchain.retrievers import (
MergerRetriever,
)
import pprint
from typing import Any, Dict
from huggingface_hub import login
login(token=os.getenv("HUGGING_FACE_TOKEN"))
def load_chunk_persist_pdf(task) -> Chroma:
pdf_folder_path = os.path.join(os.getcwd(),Path(f"data/pdf/{task}"))
documents = []
for file in os.listdir(pdf_folder_path):
if file.endswith('.pdf'):
pdf_path = os.path.join(pdf_folder_path, file)
loader = PyPDFLoader(pdf_path)
documents.extend(loader.load())
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=10)
chunked_documents = text_splitter.split_documents(documents)
os.makedirs("data/chroma_store/", exist_ok=True)
vectorstore = Chroma.from_documents(
documents=chunked_documents,
embedding=MistralAIEmbeddings(),
persist_directory= os.path.join(os.getcwd(),Path("data/chroma_store/"))
)
vectorstore.persist()
return vectorstore
df = pd.DataFrame(
{
"exercise": ["Squat","Bench Press","Lunges","Pull ups"],
"sets": [4, 4, 3, 3],
"repetitions": [10, 8, 8, 8],
"rest":["2:30","2:00","1:30","2:00"]
}
)
# parser = PandasDataFrameOutputParser(dataframe=df)
# personal_info_vectorstore = load_chunk_persist_pdf("personal_info")
# zero2hero_vectorstore = load_chunk_persist_pdf("zero2hero")
# bodyweight_vectorstore = load_chunk_persist_pdf("bodyweight")
# nutrition_vectorstore = load_chunk_persist_pdf("nutrition")
# workout_vectorstore = load_chunk_persist_pdf("workout")
# zero2hero_retriever = zero2hero_vectorstore.as_retriever()
# nutrition_retriever = nutrition_vectorstore.as_retriever()
# bodyweight_retriever = bodyweight_vectorstore.as_retriever()
# workout_retriever = workout_vectorstore.as_retriever()
# personal_info_retriever = personal_info_vectorstore.as_retriever()
llm = ChatMistralAI(model="mistral-large-latest", mistral_api_key=mistral_api_key, temperature=0)
# prompt = PromptTemplate(
# template="""
# You are a professional AI coach specialized in building fitness plans, full workout programs.
# You must adapt to the user according to personal informations in the context. A You are gentle and motivative.
# Use the following pieces of retrieved context to answer the user's query.
# Context: {context}
# \n{format_instructions}\n{question}\n
# """,
# input_variables=["question","context"],
# partial_variables={"format_instructions": parser.get_format_instructions()},
# )
# def format_docs(docs):
# return "\n\n".join(doc.page_content for doc in docs)
# def format_parser_output(parser_output: Dict[str, Any]) -> None:
# for key in parser_output.keys():
# parser_output[key] = parser_output[key].to_dict()
# return pprint.PrettyPrinter(width=4, compact=True).pprint(parser_output)
# retriever = MergerRetriever(retrievers=[zero2hero_retriever, bodyweight_retriever, nutrition_retriever, workout_retriever, personal_info_retriever])
# chain = (
# {"context": zero2hero_retriever | format_docs, "question": RunnablePassthrough()}
# | prompt
# | llm
# | parser
# )
# # chain = prompt | llm | parser
# format_parser_output(chain.invoke("Build me a full body workout plan for summer body."))
from pydantic import BaseModel, Field
from typing import List
from langchain_core.output_parsers import JsonOutputParser
class Exercise(BaseModel):
exercice: str = Field(description="Name of the exercise")
nombre_series: int = Field(description="Number of sets for the exercise")
nombre_repetitions: int = Field(description="Number of repetitions for the exercise")
temps_repos: str = Field(description="Rest time between sets")
class MusculationProgram(BaseModel):
exercises: List[Exercise]
from langchain.prompts import PromptTemplate
# Define your query to get a musculation program.
musculation_query = "Provide a musculation program with exercises, number of sets, number of repetitions, and rest time between sets."
# Set up a parser + inject instructions into the prompt template.
parser = JsonOutputParser(pydantic_object=MusculationProgram)
prompt = PromptTemplate(
template="Answer the user query.\n{format_instructions}\n{query}\n",
input_variables=["query"],
partial_variables={"format_instructions": parser.get_format_instructions()},
)
# Set up a chain to invoke the language model with the prompt and parser.
workout_chain = prompt | llm | parser
|