Spaces:
Runtime error
Runtime error
File size: 3,467 Bytes
7d7a6a9 7244d35 7d7a6a9 7244d35 7d7a6a9 e9d7936 7d7a6a9 e9d7936 7d7a6a9 b6e72f6 7d7a6a9 c525e42 7d7a6a9 c425f6c 7244d35 7d7a6a9 b6e72f6 37f14ac b6e72f6 7244d35 c525e42 7244d35 b6e72f6 c525e42 7d7a6a9 b6e72f6 7d7a6a9 b6e72f6 c425f6c 7d7a6a9 c425f6c 7d7a6a9 c425f6c 7244d35 |
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 |
from dotenv import load_dotenv
from strictjson import strict_json_async
from prompts import (
AGENT_PROMPT,
EXTRACT_SYS_PROMPT,
EXTRACT_USER_PROMPT,
RAG_SYS_PROMPT,
RAG_USER_PROMPT,
)
from sarvam import speaker, translator
from scraper import extract
load_dotenv()
async def llm(system_prompt: str, user_prompt: str) -> str:
import os
from groq import AsyncGroq
client = AsyncGroq(api_key=os.getenv("GROQ_API_KEY"))
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]
chat_completion = await client.chat.completions.create(
messages=messages,
model="llama3-70b-8192",
temperature=0.3,
max_tokens=360,
top_p=1,
stop=None,
stream=False,
)
return chat_completion.choices[0].message.content
async def call_agent(user_prompt, collection):
grade, subject, chapter = collection.split("_")
system_prompt = AGENT_PROMPT.format(grade, subject)
result = await strict_json_async(
system_prompt=system_prompt,
user_prompt=user_prompt,
output_format={
"function": 'Type of function to call, type: Enum["retriever", "translator", "speaker", "none", "extractor"]',
"keywords": "Array of keywords, type: List[str]",
"src_lang": "Identify the language that the user query is in, type: str",
"dest_lang": """Identify the target language from the user query if the function is either "translator" or "speaker". If language is not found, return "none",
type: Enum["hindi", "bengali", "kannada", "malayalam", "marathi", "odia", "punjabi", "tamil", "telugu", "english", "gujarati", "none"]""",
"source": "Identify the sentence that the user wants to translate or speak. Else return 'none', type: Optional[str]",
"url": "Identify if any URL or link is provided in the user query, type: str",
"response": "Your response, type: Optional[str]",
},
llm=llm,
)
return result
async def retriever(user_prompt, collection, client):
grade, subject, chapter = collection.split("_")
data = client.search(collection, user_prompt)
data = [i.document for i in data]
system_prompt = RAG_SYS_PROMPT.format(subject, grade)
user_prompt = RAG_USER_PROMPT.format(data, user_prompt)
return await llm(system_prompt, user_prompt)
async def extractor(user_prompt, url):
text = await extract(url)
system_prompt = EXTRACT_SYS_PROMPT.format(url)
user_prompt = EXTRACT_USER_PROMPT.format(text, user_prompt)
return await llm(system_prompt, user_prompt)
async def function_caller(user_prompt, collection, client):
result = await call_agent(user_prompt, collection)
print(f"Agent log -\n {result} \n\n")
function = result["function"].lower()
if function == "none":
return {"text": result["response"]}
elif function == "retriever":
response = await retriever(user_prompt, collection, client)
return {"text": response}
elif function == "translator":
return await translator(result["source"], result["src_lang"], result["dest_lang"])
elif function == "speaker":
return await speaker(result["source"])
elif function == "extractor":
response = await extractor(user_prompt, result["url"])
return {"text": response}
|