File size: 3,115 Bytes
2214088 |
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 |
from langchain_groq import ChatGroq
from langchain_community.utilities import GoogleSerperAPIWrapper
from src.settings import settings
from src.vectorstore import answer_query_from_existing_collection
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from src.schemas import RagResponse
class AnswerQuery:
def __init__(self, model_name: str = "llama-3.3-70b-versatile"):
"""
Class to handle the Groq model for answering queries.
"""
self.llm = ChatGroq(
model_name=model_name,
temperature=0.3,
max_tokens=512,
api_key=settings.GROQ_API_KEY,
)
self.serper = GoogleSerperAPIWrapper(serper_api_key=settings.SERPER_API_KEY)
def format_docs(self,docs):
return "\n\n".join(doc.page_content for doc in docs)
async def answer_query(
self, vectorembedding, query: str, collection_name: str = "recipe"
):
"""
Answer a query using the Groq model.
"""
vector_store = await answer_query_from_existing_collection(
vectorembedding=vectorembedding,
collection_name_=collection_name,
)
# Retriever
retriever = vector_store.as_retriever(
search_type="mmr",
search_kwargs={"k": 3, "lambda_mult": 0.5},
)
template = """
Answer using ONLY the context below:
Context: {context}
Question: {question}
If context doesn't match with the question, say,I couldn’t find information about this,and set web_search to true.
Otherwise, set web_search to false and answer only according to the context.
"""
prompt = PromptTemplate.from_template(template)
chain = (
{
"context": retriever|self.format_docs,
"question": RunnablePassthrough(),
}
| prompt
| self.llm.with_structured_output(
RagResponse,
)
)
response = chain.invoke(query)
return response
async def search_web(self, query: str):
"""Search the web for a query"""
response = self.serper.run(query)
template = """
Answer using ONLY the context below:
Context: {context}
Question: {question}
If context doesn't match with the question, say,I couldn’t find information about this.
"""
prompt = PromptTemplate.from_template(template)
chain = (
{
"context": lambda x :response,
"question": RunnablePassthrough(),
}
| prompt
| self.llm
)
response = chain.invoke(query)
return response.content
if __name__ == "__main__":
async def main():
answer_query = AnswerQuery()
query = "What is the capital of France?"
response = await answer_query.answer_query(query)
print(response)
import asyncio
asyncio.run(main())
|