123

app.py

@@ -22,6 +22,8 @@ from transformers import (
 )
 from transformers.image_utils import load_image
 
+from data_experiments import filter_experiments
+
 # Constants for text generation
 MAX_MAX_NEW_TOKENS = 2048
 DEFAULT_MAX_NEW_TOKENS = 1024
@@ -103,11 +105,16 @@ def process_image(
         yield buffer, buffer
 
 
-# Define examples for image and video inference
-image_examples = [
-    ["Jsonify Data.", "images/1.jpg"],
-    ["Explain the pie-chart in detail.", "images/2.jpg"]
-]
+experiments_to_select = filter_experiments(
+    product_id="chemistry",
+)
+experiments_to_select = sorted(experiments_to_select, key=lambda x: x["id"])
+
+def format_experiment_dropdown_title(experiment: dict) -> str:
+    id = str(experiment["id"])
+    title = experiment.get("title", "")
+    product = experiment.get("product_id", "")
+    return f"{id}. {title} ({product})"
 
 css = """
 .submit-btn {
@@ -129,12 +136,18 @@ with gr.Blocks(css=css, theme="bethecloud/storj_theme") as demo:
     gr.Markdown("# **Experiments photo moderation**")
     with gr.Row():
         with gr.Column():
+            gr.Dropdown(
+                choices=[(format_experiment_dropdown_title(exp), exp["id"]) for exp in experiments_to_select],
+                value=None,
+                label="Select Experiment",
+            )
             image_query = gr.Textbox(label="Query Input", placeholder="Enter your query here...")
             image_upload = gr.Image(type="pil", label="Image")
             image_submit = gr.Button("Submit", elem_classes="submit-btn")
-            gr.Examples(
-                examples=image_examples,
-                inputs=[image_query, image_upload]
+            model_choice = gr.Radio(
+                choices=["Qwen2.5-VL-7B-Instruct", "Qwen2.5-VL-3B-Instruct"],
+                label="Select Model",
+                value="Qwen2.5-VL-7B-Instruct"
             )
             with gr.Accordion("Advanced options", open=False):
                 max_new_tokens = gr.Slider(label="Max new tokens", minimum=1, maximum=MAX_MAX_NEW_TOKENS, step=1, value=DEFAULT_MAX_NEW_TOKENS)
@@ -150,12 +163,6 @@ with gr.Blocks(css=css, theme="bethecloud/storj_theme") as demo:
             
                 with gr.Accordion("(Result.md)", open=False):
                     markdown_output = gr.Markdown()
-                
-            model_choice = gr.Radio(
-                choices=["Qwen2.5-VL-7B-Instruct", "Qwen2.5-VL-3B-Instruct"],
-                label="Select Model",
-                value="Qwen2.5-VL-7B-Instruct"
-            )
             
     image_submit.click(
         fn=process_image,

data.py

@@ -0,0 +1,11 @@
+import json
+
+def load_jsonl(file_path) -> list[dict]:
+    result = []
+    with open(file_path, 'r', encoding='utf-8') as file:
+        for line in file:
+            if not line.strip() or line.startswith('#') or line.startswith('//'):
+                continue
+            json_line = json.loads(line)
+            result.append(json_line)
+    return result

data_experiments.py

@@ -0,0 +1,55 @@
+from typing import Union
+from collections import defaultdict
+
+from data import load_jsonl
+
+
+all_experiments = load_jsonl('experiments.jsonl')
+
+experiment_ids_by_product_id = defaultdict(list)
+for experiment in all_experiments:
+    product_id = experiment.get('product_id')
+    if product_id is not None:
+        experiment_ids_by_product_id[product_id].append(experiment["id"])
+
+
+def get_experiment_by_id(experiment_id: int) -> Union[dict, None]:
+    for experiment in all_experiments:
+        if experiment['id'] == experiment_id:
+            return experiment
+    return None
+
+
+def filter_experiments(
+    experiments: Union[list[dict], None] = None,
+    product_id: Union[str, list[str], None] = None,
+    experiment_id: Union[int, list[int], None] = None,
+) -> list[dict]:
+    filtered_experiments = experiments
+    if filtered_experiments is None:
+        filtered_experiments = all_experiments
+    if product_id is not None:
+        if isinstance(product_id, str):
+            product_id = {product_id}
+        filtered_experiments = [exp for exp in filtered_experiments if exp.get('product_id') in product_id]
+    if experiment_id is not None:
+        if isinstance(experiment_id, int):
+            experiment_id = {experiment_id}
+        filtered_experiments = [exp for exp in filtered_experiments if exp['id'] in experiment_id]
+    return filtered_experiments
+
+
+def get_experiment(
+    experiments: Union[list[dict], None] = None,
+    product_id: Union[str, list[str], None] = None,
+    experiment_id: Union[int, None] = None,
+) -> Union[dict, None]:
+    filtered_experiments = filter_experiments(
+        experiments=experiments,
+        product_id=product_id,
+        experiment_id=experiment_id
+    )
+    if filtered_experiments:
+        return filtered_experiments[0]
+    else:
+        return None

data_photos.py

@@ -0,0 +1,87 @@
+from typing import Union
+from enum import Enum
+
+from data import load_jsonl
+from data_experiments import experiment_ids_by_product_id
+
+class PhotoModerationState(Enum):
+    INAPPROPRIATE = "inappropriate"
+    SAFE = "safe"
+    SAFE_FOR_SHARING = "safe_for_sharing"
+    UNKNOWN = "unknown"
+
+class PhotoApprovalState(Enum):
+    APPROVED = "approved"
+    FAILED = "failed"
+    SUPERSEDED = "superseded"
+    UNKNOWN = "unknown"
+
+all_photos = load_jsonl('mobile_photo_uploads_2.jsonl')
+
+def get_photo_by_id(photo_id: str) -> Union[dict, None]:
+    for photo in all_photos:
+        if photo['uid'] == photo_id:
+            return photo
+    return None
+
+def get_photo_url(photo: dict) -> str:
+    uid = photo['uid']
+    url = f'https://res.cloudinary.com/mel-science/image/upload/c_limit,w_1024,h_1024/v1/dotcomprodmobup/{uid}.jpg'
+    return url
+
+def filter_photos(
+    photos: Union[list[dict], None] = None, 
+    experiment_id: Union[int, list[int], None] = None,
+    product_id: Union[str, list[str], None] = None,
+    uid: Union[str, set[str], None] = None,
+    is_sharing_requested: Union[bool, None] = None,
+    moderation_state: Union[set[PhotoModerationState], None] = None,
+    approval_state: Union[set[PhotoApprovalState], None] = None,
+) -> list[dict]:
+    filtered_photos = photos
+    if filtered_photos is None:
+        filtered_photos = all_photos
+    if experiment_id is not None:
+        if isinstance(experiment_id, int):
+            experiment_id = {experiment_id}
+        filtered_photos = [photo for photo in filtered_photos if int(photo.get('experiment_id')) in experiment_id]
+    if product_id is not None:
+        if isinstance(product_id, str):
+            product_id = {product_id}
+        product_experiment_ids = set()
+        for id in product_id:
+            product_experiment_ids.update(experiment_ids_by_product_id.get(id, []))
+        filtered_photos = [photo for photo in filtered_photos if int(photo.get('experiment_id')) in product_experiment_ids]
+    if uid is not None:
+        if isinstance(uid, str):
+            uid = {uid}
+        filtered_photos = [photo for photo in filtered_photos if photo.get('uid') in uid]
+    if is_sharing_requested is not None:
+        filtered_photos = [photo for photo in filtered_photos if photo.get('is_sharing_requested') == str(is_sharing_requested)]
+    if moderation_state is not None:
+        values = {state.value for state in moderation_state}
+        filtered_photos = [photo for photo in filtered_photos if photo.get('moderation_state') in values]
+    if approval_state is not None:
+        values = {state.value for state in approval_state}
+        filtered_photos = [photo for photo in filtered_photos if photo.get('approval_state') in values]
+    return filtered_photos
+
+def get_photo(
+    photos: Union[list[dict], None] = None, 
+    experiment_id: Union[int, None] = None,
+    uid: Union[str, set[str], None] = None,
+    is_sharing_requested: Union[bool, None] = None,
+    moderation_state: Union[set[PhotoModerationState], None] = None,
+    approval_state: Union[set[PhotoApprovalState], None] = None,
+) -> Union[dict, None]:
+    filtered_photos = filter_photos(
+        photos=photos,
+        experiment_id=experiment_id,
+        uid=uid,
+        is_sharing_requested=is_sharing_requested,
+        moderation_state=moderation_state,
+        approval_state=approval_state,
+    )
+    if not filtered_photos:
+        return None
+    return filtered_photos[0]

llm_messages_v1.py

@@ -0,0 +1,75 @@
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
+
+from data import get_photo_url
+
+def format_experiment(experiment: dict) -> str:
+    lines = []
+
+    lines.append(f"Experiment: {experiment['title']}")
+    lines.append(f"Description: {experiment['short_description']}")
+
+    experiment_steps = experiment.get('steps', [])
+    if experiment_steps:
+        lines.append("Steps:")
+        step_index = 1
+        for step in experiment_steps:
+            step_text = step.get('text', '').strip()
+            if step_text:
+                lines.append(f"  {step_index}. {step_text}")
+                step_index += 1
+
+    experiment_result = experiment.get('expected_result', "").strip()
+    if experiment_result:
+        lines.append(f"Expected result: {experiment_result}")
+
+    combined = "\n".join(lines)
+    return combined
+
+
+def format_system_message(experiment: dict) -> SystemMessage:
+    lines = []
+    lines.append("You are a photo moderation assistant for science experiments.")
+    lines.append("Your task is to determine if the provided photo is a succeful result of the experiment, completed setup or user is on the right way to result. And doesn't contain any inappropriate content. Or not.")
+    lines.append("")
+    lines.append("Here is the experiment details:")
+    experiment_text = format_experiment(experiment)
+    lines.append(experiment_text)
+
+    combined = "\n".join(lines)
+    return SystemMessage(content=combined)
+
+def format_human_message(experiment: dict, photo: dict) -> HumanMessage:
+    lines = []
+    lines.append("Here is the photo to check")
+    # lines.append("Please analyze the photo and determine if it is a successful result of the experiment, completed setup or user is on the right way to result. And doesn't contain any inappropriate content.")
+    lines.append("Result options:")
+    lines.append("- `approved`: photo is a successful result of the experiment")
+    lines.append("- `partial`: photo is something closed to the experiment, but it is not a final successful result (e.g. not all steps are completed, or parts of result are not visible)")
+    # lines.append("- `inappropriate`: photo contains inappropriate content (including faces of children)")
+    lines.append("- `failed`: photo is not a successful result of the experiment, but doesn't contain inappropriate content")
+    # lines.append("Please analyze the photo and determine if it is a successful result of the experiment.")
+    lines.append("Answer in the following format:")
+    lines.append("```json")
+    lines.append('{"result": "approved" | "partial" | "failed", "reason": "your reason for the decision"}')
+    lines.append("```")
+    combined = "\n".join(lines)
+    photo_url = get_photo_url(photo)
+    return HumanMessage(content=[
+        {"type": "text", "text": combined},
+        {"type": "image_url", "image_url": {"url": photo_url}}
+    ])
+
+def format_ai_message(experiment: dict, photo: dict) -> AIMessage:
+    is_approved = photo.get('approval_state', 'unknown') == 'approved'
+    result = "approved" if is_approved else "failed"
+    reason = photo.get('review_comment', '').strip()
+
+    lines = []
+    lines.append("```json")
+    lines.append("{" + f'"result": "{result}", "reason": "{reason}"' + "}")
+    lines.append("```")
+    combined = "\n".join(lines)
+
+    return AIMessage(content=[
+        {"type": "text", "text": combined},
+    ])

llm_messages_v2.py

@@ -0,0 +1,68 @@
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
+
+from data import get_photo_url
+
+def format_experiment(experiment: dict) -> str:
+    lines = []
+
+    lines.append(f"Experiment: {experiment['title']}")
+    lines.append(f"Description: {experiment['short_description']}")
+
+    experiment_steps = experiment.get('steps', [])
+    if experiment_steps:
+        lines.append("Steps:")
+        step_index = 1
+        for step in experiment_steps:
+            step_text = step.get('text', '').strip()
+            if step_text:
+                lines.append(f"  {step_index}. {step_text}")
+                step_index += 1
+
+    experiment_result = experiment.get('expected_result', "").strip()
+    if experiment_result:
+        lines.append(f"Expected result: {experiment_result}")
+
+    combined = "\n".join(lines)
+    return combined
+
+
+def format_system_message(experiment: dict) -> SystemMessage:
+    lines = []
+    lines.append("You are a photo moderation assistant for science experiments.")
+    lines.append("Your task is to determine if the provided photo is a succeful result of the experiment, completed setup or user is on the right way to result. And doesn't contain any inappropriate content. Or not.")
+    # lines.append("")
+    # lines.append("Here is the experiment details:")
+    # experiment_text = format_experiment(experiment)
+    # lines.append(experiment_text)
+
+    combined = "\n".join(lines)
+    return SystemMessage(content=combined)
+
+def format_human_message(experiment: dict, photo: dict) -> HumanMessage:
+    lines = []
+    # lines.append("Here is the photo to check")
+    # lines.append("Please analyze the photo and determine if it is a successful result of the experiment.")
+    lines.append("Answer in the following format:")
+    lines.append("```json")
+    lines.append('{"result": "approved" | "failed"}')
+    lines.append("```")
+    combined = "\n".join(lines)
+    photo_url = get_photo_url(photo)
+    return HumanMessage(content=[
+        {"type": "text", "text": combined},
+        {"type": "image_url", "image_url": {"url": photo_url}}
+    ])
+
+def format_ai_message(experiment: dict, photo: dict) -> AIMessage:
+    is_approved = photo.get('approval_state', 'unknown') == 'approved'
+    result = "approved" if is_approved else "failed"
+
+    lines = []
+    lines.append("```json")
+    lines.append("{" + f'"result": "{result}"' + "}")
+    lines.append("```")
+    combined = "\n".join(lines)
+
+    return AIMessage(content=[
+        {"type": "text", "text": combined},
+    ])

llm_messages_v3.py

@@ -0,0 +1,73 @@
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
+
+from data import get_photo_url
+
+def format_experiment(experiment: dict) -> str:
+    lines = []
+
+    lines.append(f"Experiment: {experiment['title']}")
+    lines.append(f"Description: {experiment['short_description']}")
+
+    experiment_steps = experiment.get('steps', [])
+    if experiment_steps:
+        lines.append("Steps:")
+        step_index = 1
+        for step in experiment_steps:
+            step_text = step.get('text', '').strip()
+            if step_text:
+                lines.append(f"  {step_index}. {step_text}")
+                step_index += 1
+
+    experiment_result = experiment.get('expected_result', "").strip()
+    if experiment_result:
+        lines.append(f"Expected result: {experiment_result}")
+
+    combined = "\n".join(lines)
+    return combined
+
+
+def format_system_message(experiment: dict) -> SystemMessage:
+    lines = []
+    lines.append("You are a photo moderation assistant for science experiments.")
+    lines.append("Your task is to determine if the provided photo is a succeful result of the experiment, completed setup or user is on the right way to result. And doesn't contain any inappropriate content. Or not.")
+    # lines.append("")
+    # lines.append("Here is the experiment details:")
+    # experiment_text = format_experiment(experiment)
+    # lines.append(experiment_text)
+
+    combined = "\n".join(lines)
+    return SystemMessage(content=combined)
+
+def format_human_message(experiment: dict, photo: dict) -> HumanMessage:
+    lines = []
+
+    lines.append("Here is the experiment details:")
+    experiment_text = format_experiment(experiment)
+    lines.append(experiment_text)
+    lines.append("")
+
+    lines.append("Your task is to determine if the provided photo is a succeful result of the experiment, completed setup or user is on the right way to result. And doesn't contain any inappropriate content. Or not.")
+    lines.append("Answer in the following format:")
+    lines.append("```json")
+    lines.append('{"result": "approved" | "failed"}')
+    lines.append("```")
+    combined = "\n".join(lines)
+    photo_url = get_photo_url(photo)
+    return HumanMessage(content=[
+        {"type": "text", "text": combined},
+        {"type": "image_url", "image_url": {"url": photo_url}}
+    ])
+
+def format_ai_message(experiment: dict, photo: dict) -> AIMessage:
+    is_approved = photo.get('approval_state', 'unknown') == 'approved'
+    result = "approved" if is_approved else "failed"
+
+    lines = []
+    lines.append("```json")
+    lines.append("{" + f'"result": "{result}"' + "}")
+    lines.append("```")
+    combined = "\n".join(lines)
+
+    return AIMessage(content=[
+        {"type": "text", "text": combined},
+    ])