Spaces:
Sleeping
Sleeping
File size: 3,082 Bytes
fa8520f 6874dac 38cf703 6c655a3 fa8520f 38cf703 fa8520f 38cf703 fa8520f 38cf703 fa8520f 38cf703 5c271a3 fa8520f 38cf703 6874dac 38cf703 6874dac 38cf703 6874dac 38cf703 6c655a3 8ce97f0 |
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 |
import pandas as pd
import ast
from groq import Groq
import os
from .prompts import captioning_prompt
from langchain_core.messages import FunctionMessage , AIMessage
from .tools import InfluencerRetrievalTool
import re
class ImageCaptioner:
def __init__(self, api_key=os.environ.get('GROQ_API_KEY')):
self.client = Groq(api_key=api_key)
def caption_image(self,image_base64,user_input):
if len(image_base64)>0:
print('Captioning image')
chat_completion = self.client.chat.completions.create(
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": captioning_prompt(user_input)},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpg;base64,{image_base64[-1]}",
},
},
],
}
],
model="meta-llama/llama-4-scout-17b-16e-instruct",
max_completion_tokens=50,
temperature = 1
)
response=chat_completion.choices[0].message.content
return response
else:
return ''
class AnalyticsViewer:
def __init__(self, business_details):
self.business_details = business_details
def show_analytics(self):
tool_response = InfluencerRetrievalTool().retrieve_for_analytics(str(self.business_details))
return tool_response
class ResponseBlockExtractor:
def __init__(self, response):
self.response = response
def extract_latest(self):
latest_block = []
temp_block = []
# Reverse iterate through the messages
for message in reversed(self.response):
if isinstance(message, (FunctionMessage, AIMessage)):
temp_block.insert(0, message.content)
# Once we collect 3 items in correct structure, stop
if len(temp_block) == 3:
if "tool=" in temp_block[1] and "query_response" in temp_block[1]:
latest_block = temp_block
break
else:
temp_block = []
print('The latest block', latest_block)
return latest_block
def handle_tools(tools_list,stored_data):
tools_order = [
"analytics",
"ideation",
"human-idea-refining",
"generate-story",
"generate-ultimate-story",
"generate-image",
]
if 'generate-story' in tools_list and len(stored_data['human_ideation_interactions'])<1:
tools_list.append('human-idea-refining')
if 'generate-ultimate-story' in tools_list and len(stored_data['brainstorming_response'])<1:
tools_list.append('generate-story')
tools_list = [tool for tool in tools_order if tool in tools_list]
return tools_list
|