Spaces:
Running
Running
File size: 3,142 Bytes
96f6720 |
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 |
import base64
from langchain_core.messages import HumanMessage
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from image_generator import generate_image
from llm import get_text_llm, get_vision_llm, invoke_llm_async
from time_it import time_it_async
from util import UTF8_ENCODING, load_prompt
async def get_recipe_from_image(image_path: str) -> dict:
recipe: dict = await _analyze_image_from_path(image_path, 'get_recipe_from_image.prompt.txt')
return recipe
async def get_altered_recipe(orig_recipe: dict, restrictions: list[str], diagnoses: list[str]) -> dict:
recipe: dict = await _generate_text('alter_recipe_for_healthier.prompt.txt',
{'recipe': orig_recipe, 'dietary_restrictions': restrictions, 'medical_diagnoses': diagnoses})
if 'recipe' in recipe:
recipe = recipe['recipe']
return recipe
async def _analyze_image_from_path(image_path: str, prompt_file: str) -> dict | str:
with open(image_path, 'rb') as image_file:
image_data = image_file.read()
image_format = _get_image_format(image_path)
return await _analyze_image_from_bytes(image_data, prompt_file, image_format)
@time_it_async
async def _analyze_image_from_bytes(image_data: bytes, prompt_file: str, image_format: str, output_parser_type=JsonOutputParser) -> dict | str:
img_base64 = base64.b64encode(image_data).decode(UTF8_ENCODING)
prompt = load_prompt(prompt_file)
prompt_template = ChatPromptTemplate.from_messages([
HumanMessage(
content=[
{'type': 'text', 'text': prompt},
{
'type': 'image_url',
'image_url': {
'url': f'data:image/{image_format};base64,{img_base64}',
'detail': 'high'
}
}
]
),
])
llm = get_vision_llm()
chain = prompt_template | llm | output_parser_type()
response = await invoke_llm_async(chain)
return response
def _get_image_format(image_path: str) -> str:
file_ext = image_path.split('.')[-1].lower()
match file_ext:
case 'jpg' | 'jpeg':
return 'jpeg'
case 'png' | 'gif' | 'webp':
return file_ext
case _:
raise ValueError(f'Unsupported image format for {image_path=}')
async def _generate_text(prompt_file: str, input: dict) -> dict | str:
prompt = load_prompt(prompt_file)
prompt_template = PromptTemplate.from_template(prompt)
llm = get_text_llm()
chain = prompt_template | llm | JsonOutputParser()
response = await invoke_llm_async(chain, input)
return response
def get_image_from_recipe(recipe: dict) -> str:
recipe_for_image_gen = {k: v for k, v in recipe.items() if k in {'name', 'ingredients', 'instructions', 'meal_type', 'serves'}}
image_url = generate_image('get_image_from_recipe.prompt.txt', {'recipe': recipe_for_image_gen})
return image_url
|