# Import dependencies
from encord.user_client import EncordUserClient
from encord.workflow import AgentStage
import openai
import base64
import requests
import json
# Initialise your OpenAI client
openai.api_key = "<your_openai_api_key>"
def get_classification_from_the_model(media_content):
"""
Example function that passes media to OpenAI's ChatGPT API along with the prompt
and parses the result.
"""
prompt = """
You are an image analysis expert. You're working on a project that includes annotation of different pets images.
Your task is to assign one of the following tags to the image: "cat", "dog", "other".
Reply in JSON format of the following structure: { "classification": cat|dog|other }
"""
completion = openai.ChatCompletion.create(
model="gpt-4o-mini",
messages=[
ChatCompletionSystemMessageParam(role="system", content=prompt),
ChatCompletionUserMessageParam(
role="user",
content=[
ChatCompletionContentPartImageParam(
image_url=ImageURL(url=f"data:image/jpeg;base64,{media_content}", detail="auto"),
type="image_url",
)
]
),
],
response_format=ResponseFormat(type="json_object"),
max_tokens=1000,
)
raw_text_completion = completion.choices[0].message.content
try:
parsed_result = json.loads(raw_text_completion)
return parsed_result["classification"].lower()
except Exception as e:
print(f"Failed to process the model response: {e}")
return None
# Authenticate using the path to your private key
user_client = EncordUserClient.create_with_ssh_private_key(
ssh_private_key_path="<private_key_path>"
)
# Specify the Project that contains the Task agent. Replace <project_hash> with the hash of your Project
project = user_client.get_project("<project_hash>")
radio_classification = project.ontology_structure.get_child_by_title(
title="Animal",
type_=Classification,
)
cat_option = radio_ontology_classification.get_child_by_title(
title="Cat", type_=Option
)
dog_option = radio_ontology_classification.get_child_by_title(
title="Dog", type_=Option
)
# Specify the Task Agent
agent_stage = project.workflow.get_stage(name="Agent 1", type_=AgentStage)
for task in agent_stage.get_tasks():
# Got a task for the following data unit
print(f"{task.data_hash} -> {task.data_title}")
# Getting a label row for the data unit
label_row = project.list_label_rows_v2(data_hashes=[task.data_hash])[0]
label_row.initialise_labels(include_signed_url=True)
# Downloading the media:
media_response = requests.get(label_row.data_link)
media_content = base64.b64encode(media_response.content).decode("utf-8")
# Now we can send the media to OpenAI:
model_response = get_classification_from_the_model(media_content)
# And interpret the result:
match model_response:
case "cat":
# Create a classification instance
classification_instance = (
radio_ontology_classification.create_instance()
)
radio_classification_instance.set_answer(
answer=cat_option
)
label_row.add_classification_instance(radio_classification_instance)
label_row.save()
task.proceed(pathway_name="Cat")
case "dog":
# Create & save classification instance
classification_instance = (
radio_ontology_classification.create_instance()
)
radio_classification_instance.set_answer(
answer=dog_option
)
label_row.add_classification_instance(radio_classification_instance)
label_row.save()
task.proceed(pathway_name="Dog")
case _:
task.proceed(pathway_name="Other")