Editor Agent Examples

functions-framework --target=handle_object_hashes --debug --source agent.py

source venv/bin/activate
encord-agents test local agent '<your_url>'

python -m venv venv                 # Create a virtual Python environment  
source venv/bin/activate            # Activate the virtual environment  
python -m pip install encord-agents anthropic  # Install required dependencies  
export ANTHROPIC_API_KEY="<your_api_key>"     # Set your Anthropic API key  
export ENCORD_SSH_KEY_FILE="/path/to/your/private/key"  # Define your Encord SSH key  

# 1. Import dependencies, authenticate with Encord, and set up the Project. Ensure you insert your Project's unique identifier.
import os

from anthropic import Anthropic
from encord.objects.ontology_labels_impl import LabelRowV2
from numpy.typing import NDArray
from typing_extensions import Annotated

from encord_agents.core.ontology import OntologyDataModel
from encord_agents.core.utils import get_user_client
from encord_agents.core.video import Frame
from encord_agents.gcp import Depends, editor_agent
from encord_agents.gcp.dependencies import FrameData, dep_single_frame

client = get_user_client()
project = client.get_project("<your_project_hash>")

# 2. Create a data model and a system prompt based on the Project Ontology to tell Claude how to structure its response
data_model = OntologyDataModel(project.ontology_structure.classifications)

system_prompt = f"""
You're a helpful assistant that's supposed to help fill in json objects 
according to this schema:

    ```json
    {data_model.model_json_schema_str}
    ```

Please only respond with valid json.
"""

# 3. Set up an Anthropic API client to establish communication with Claude 
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
anthropic_client = Anthropic(api_key=ANTHROPIC_API_KEY)

# 4. Define the Editor Agent
@editor_agent()
def agent(
    frame_data: FrameData,
    lr: LabelRowV2,
    content: Annotated[NDArray, Depends(dep_single_frame)],
):
    # # Retrieving Frame Content: It automatically fetches the current frame's image data using the `dep_single_frame` dependency
    frame = Frame(frame_data.frame, content=content)
    # Analyzing with Claude: The frame image is then sent to the Claude AI model for analysis
    message = anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=system_prompt,
        messages=[
            {
                "role": "user",
                "content": [frame.b64_encoding(output_format="anthropic")],
            }
        ],
    )
    try:
        # Parsing Classifications: Claude's response is parsed and transformed into structured classification instances using the predefined data model
        classifications = data_model(message.content[0].text)
        for clf in classifications:
            clf.set_for_frames(frame_data.frame, confidence=0.5, manual_annotation=False)
            lr.add_classification_instance(clf)
    except Exception:
        import traceback

        traceback.print_exc()
        print(f"Response from model: {message.content[0].text}")
        
    # Saving Results: The new classifications are added to the active label row, and the updated results are saved within the Project.
    lr.save()

functions-framework --target=agent --debug --source agent.py

source venv/bin/activate
encord-agents test local agent '<your_url>'

python -m venv venv                 # Create a virtual Python environment  
source venv/bin/activate            # Activate the virtual environment  
python -m pip install encord-agents anthropic  # Install required dependencies  
export ANTHROPIC_API_KEY="<your_api_key>"     # Set your Anthropic API key  
export ENCORD_SSH_KEY_FILE="/path/to/your/private/key"  # Define your Encord SSH key  

# 1. Import dependencies, authenticate with Encord, and set up the Project. Ensure you insert your Project's unique identifier
import os

from anthropic import Anthropic
from encord.objects.ontology_labels_impl import LabelRowV2
from typing_extensions import Annotated

from encord_agents.core.ontology import OntologyDataModel
from encord_agents.core.utils import get_user_client
from encord_agents.gcp import Depends, editor_agent
from encord_agents.gcp.dependencies import FrameData, InstanceCrop, dep_object_crops

# User client
client = get_user_client()
project = client.get_project("<project_hash>")

# 2. Extract the generic Ontology object and the specific objects of interest. This example sorts Ontology objects based on whether their title is `"generic"`
generic_ont_obj, *other_objects = sorted(
    project.ontology_structure.objects,
    key=lambda o: o.title.lower() == "generic",
    reverse=True,
)

# 3. Prepare the system prompt for each object crop using the `data_model` to generate the JSON schema
data_model = OntologyDataModel(other_objects)
system_prompt = f"""
You're a helpful assistant that's supposed to help fill in 
json objects according to this schema:

`{data_model.model_json_schema_str}`

Please only respond with valid json.
"""

# 4. Set up an Anthropic API client to establish communication with the Claude model. You must include your Anthropic API key

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
anthropic_client = Anthropic(api_key=ANTHROPIC_API_KEY)


# 5. Define the Editor Agent
@editor_agent()
def agent(
    frame_data: FrameData,
    lr: LabelRowV2,
    crops: Annotated[
        list[InstanceCrop],
        Depends(dep_object_crops(filter_ontology_objects=[generic_ont_obj])),
    ],
):
    # 6. Query Claude using the image crops. The `crop` variable has a convenient `b64_encoding` method to produce an input that Claude understands.
    changes = False
    for crop in crops:
        message = anthropic_client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            system=system_prompt,
            messages=[
                {
                    "role": "user",
                    "content": [crop.b64_encoding(output_format="anthropic")],
                }
            ],
        )

        # 7. Parse Claude's message using the `data_model`.
        try:
            instance = data_model(message.content[0].text)

            coordinates = crop.instance.get_annotation(frame=frame_data.frame).coordinates
            instance.set_for_frames(
                coordinates=coordinates,
                frames=frame_data.frame,
                confidence=0.5,
                manual_annotation=False,
            )
            lr.remove_object(crop.instance)
            lr.add_object_instance(instance)
            changes = True
        except Exception:
            import traceback

            traceback.print_exc()
            print(f"Response from model: {message.content[0].text}")

    # 8. Save the labels with Encord.
    if changes:
        lr.save()

functions-framework --target=agent --debug --source agent.py

source venv/bin/activate
encord-agents test local agent <your_url>

uvicorn main:app --reload --port 8080

source venv/bin/activate
encord-agents test local agent '<your_url>'

python -m venv venv                   # Create a virtual Python environment  
source venv/bin/activate              # Activate the virtual environment 
python -m pip install "fastapi[standard]" encord-agents anthropic # Install required dependencies  
export ANTHROPIC_API_KEY="<your_api_key>" # Set your Anthropic API key 
export ENCORD_SSH_KEY_FILE="/path/to/your/private/key"  # Define your Encord SSH key 

# 1. Import dependencies and set up the Project. The CORS middleware is crucial as it allows the Encord platform to make requests to your API.
import os

import numpy as np
from anthropic import Anthropic
from encord.objects.ontology_labels_impl import LabelRowV2
from fastapi import Depends
from numpy.typing import NDArray
from typing_extensions import Annotated

from encord_agents.core.data_model import Frame
from encord_agents.core.ontology import OntologyDataModel
from encord_agents.core.utils import get_user_client
from encord_agents.fastapi.cors import get_encord_app
from encord_agents.fastapi.dependencies import (
    FrameData,
    dep_label_row,
    dep_single_frame,
)

# Initialize FastAPI app
app = get_encord_app()

# 2. Set up the Project and create a data model based on the Ontology.

client = get_user_client()
project = client.get_project("<your_project_hash>")
data_model = OntologyDataModel(project.ontology_structure.classifications)

# 3. Set up Claude and create the system prompt that tells Claude how to structure its response.
system_prompt = f"""
You're a helpful assistant that's supposed to help fill in json objects 
according to this schema:

    ```json
    {data_model.model_json_schema_str}
    ```

Please only respond with valid json.
"""

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
anthropic_client = Anthropic(api_key=ANTHROPIC_API_KEY)

# 4. Define the Editor Agent
@app.post("/frame_classification")
async def classify_frame(
    frame_data: FrameData,
    lr: Annotated[LabelRowV2, Depends(dep_label_row)],
    content: Annotated[NDArray[np.uint8], Depends(dep_single_frame)],
):
    # Receives frame data using FastAPI's Form dependency.
    # Note: FastAPI handles parsing the incoming request body (which implicitly includes frame_data,
    # and the dependencies (dep_label_row, dep_single_frame) resolve the lr and content).

    """Classify a frame using Claude."""
    # Constructs a `Frame` object with the content.
    frame = Frame(frame=frame_data.frame, content=content) 
    
    # Sends the frame image to Claude for analysis.
    message = anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=system_prompt,
        messages=[
            {
                "role": "user",
                "content": [frame.b64_encoding(output_format="anthropic")],
            }
        ],
    )
    try:
        # Parses Claude's response into classification instances.
        classifications = data_model(message.content[0].text) 
        for clf in classifications:
            clf.set_for_frames(frame_data.frame, confidence=0.5, manual_annotation=False)
            # Adds classifications to the label row
            lr.add_classification_instance(clf) 
    except Exception:
        import traceback

        traceback.print_exc()
        print(f"Response from model: {message.content[0].text}")

    # Saves the updated data.
    lr.save()

uvicorn main:app --reload --port 8080

source venv/bin/activate
encord-agents test local frame_classification '<your_url>'

python -m venv venv                 # Create a virtual Python environment  
source venv/bin/activate            # Activate the virtual environment  
python -m pip install encord-agents anthropic  # Install required dependencies  
export ANTHROPIC_API_KEY="<your_api_key>"     # Set your Anthropic API key  
export ENCORD_SSH_KEY_FILE="/path/to/your/private/key"  # Define your Encord SSH key  

# 1. Import dependencies and set up the Project. The CORS middleware is crucial as it allows the Encord platform to make requests to your API.
import os

import numpy as np
from anthropic import Anthropic
from encord.objects.ontology_labels_impl import LabelRowV2
from fastapi import Depends
from numpy.typing import NDArray
from typing_extensions import Annotated

from encord_agents.core.data_model import Frame
from encord_agents.core.ontology import OntologyDataModel
from encord_agents.core.utils import get_user_client
from encord_agents.fastapi.cors import get_encord_app
from encord_agents.fastapi.dependencies import (
    FrameData,
    dep_label_row,
    dep_single_frame,
)

# Initialize FastAPI app
app = get_encord_app()

# 2. Set up the Project and create a data model based on the Ontology.

client = get_user_client()
project = client.get_project("<your_project_hash>")
data_model = OntologyDataModel(project.ontology_structure.classifications)

# 3. Set up Claude and create the system prompt that tells Claude how to structure its response.
system_prompt = f"""
You're a helpful assistant that's supposed to help fill in json objects 
according to this schema:

    ```json
    {data_model.model_json_schema_str}
    ```

Please only respond with valid json.
"""

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
anthropic_client = Anthropic(api_key=ANTHROPIC_API_KEY)

# 4. Define the Editor Agent
@app.post("/frame_classification")
async def classify_frame(
    frame_data: FrameData,
    lr: Annotated[LabelRowV2, Depends(dep_label_row)],
    content: Annotated[NDArray[np.uint8], Depends(dep_single_frame)],
):
    # Receives frame data using FastAPI's Form dependency.
    # Note: FastAPI handles parsing the incoming request body (which implicitly includes frame_data,
    # and the dependencies (dep_label_row, dep_single_frame) resolve the lr and content).

    """Classify a frame using Claude."""
    # Constructs a `Frame` object with the content.
    frame = Frame(frame=frame_data.frame, content=content) 
    
    # Sends the frame image to Claude for analysis.
    message = anthropic_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=system_prompt,
        messages=[
            {
                "role": "user",
                "content": [frame.b64_encoding(output_format="anthropic")],
            }
        ],
    )
    try:
        # Parses Claude's response into classification instances.
        classifications = data_model(message.content[0].text) 
        for clf in classifications:
            clf.set_for_frames(frame_data.frame, confidence=0.5, manual_annotation=False)
            # Adds classifications to the label row
            lr.add_classification_instance(clf) 
    except Exception:
        import traceback

        traceback.print_exc()
        print(f"Response from model: {message.content[0].text}")

    # Saves the updated data.
    lr.save()

fastapi dev agent.py --port 8080

source venv/bin/activate
encord-agents test local object_classification '<your_url>'

python -m venv venv                 # Create a virtual Python environment  
source venv/bin/activate            # Activate the virtual environment  
python -m pip install encord-agents langchain-openai "fastapi[standard]" openai  # Install required dependencies  
export OPENAI_API_KEY="<your-api-key>"     # Set your OpenAI API key  
export ENCORD_SSH_KEY_FILE="/path/to/your/private/key"  # Define your Encord SSH key  

# 1. Set up imports and create a Pydantic model for our LLM's structured output.
import os
from typing import Annotated

import numpy as np
from encord.exceptions import LabelRowError
from encord.objects.classification_instance import ClassificationInstance
from encord.objects.ontology_labels_impl import LabelRowV2
from fastapi import Depends
from langchain_openai import ChatOpenAI
from numpy.typing import NDArray
from pydantic import BaseModel

from encord_agents import FrameData
from encord_agents.fastapi.cors import get_encord_app
from encord_agents.fastapi.dependencies import Frame, dep_label_row, dep_single_frame

# The response model for the agent to follow.
class AgentCaptionResponse(BaseModel):
    rephrase_1: str
    rephrase_2: str
    rephrase_3: str


# 2. Create a detailed system prompt for the LLM that explains exactly what kind of rephrasing we want.
SYSTEM_PROMPT = """
You are a helpful assistant that rephrases captions.

I will provide you with a video caption and an image of the scene of the video. 

The captions follow this format:

"The droid picks up <cup_0> and puts it on the <table_0>."

The captions that you make should replace the tags, e.g., <cup_0>, with the actual object names.
The replacements should be consistent with the scene.

Here are three rephrases: 

1. The droid picks up the blue mug and puts it on the left side of the table.
2. The droid picks up the cup and puts it to the left of the plate.
3. The droid is picking up the mug on the right side of the table and putting it down next to the plate.

You will rephrase the caption in three different ways, as above, the rephrases should be

1. Diverse in terms of adjectives, object relations, and object positions.
2. Sound in relation to the scene. You cannot talk about objects you cannot see.
3. Short and concise. Keep it within one sentence.

"""

# 3. Configure the LLM to use structured outputs based on our model.
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.4, api_key=os.environ["OPENAI_API_KEY"])
llm_structured = llm.with_structured_output(AgentCaptionResponse)


# 4. Create a helper function to prompt the model with both text and image.
def prompt_gpt(caption: str, image: Frame) -> AgentCaptionResponse:
    prompt = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {
            "role": "user",
            "content": [
                {"type": "text", "text": f"Video caption: `{caption}`"},
                image.b64_encoding(output_format="openai"),
            ],
        },
    ]
    return llm_structured.invoke(prompt)


# 5. Initialize the FastAPI app with the required CORS middleware.
app = get_encord_app()


# 6. Define the agent to handle the recaptioning.
@app.post("/my_agent")
def my_agent(
    frame_data: FrameData,
    label_row: Annotated[LabelRowV2, Depends(dep_label_row)],
    frame_content: Annotated[NDArray[np.uint8], Depends(dep_single_frame)],
) -> None:
    # Get the relevant Ontology information
    # Recall that we expect
    # [human annotation, llm recaption 1, llm recaption 2, llm recaption 3]
    # in the Ontology
    cap, *rs = label_row.ontology_structure.classifications

    # Retrieve the existing human-created caption, prioritizing captions from the current frame or falling back to frame zero.
    instances = label_row.get_classification_instances(
        filter_ontology_classification=cap, filter_frames=[0, frame_data.frame]
    )
    if not instances:
        # nothing to do if there are no human labels
        return
    elif len(instances) > 1:

        def order_by_current_frame_else_frame_0(
            instance: ClassificationInstance,
        ) -> bool:
            try:
                instance.get_annotation(frame_data.frame)
                return 2  # The best option
            except LabelRowError:
                pass
            try:
                instance.get_annotation(0)
                return 1
            except LabelRowError:
                return 0

        instance = sorted(instances, key=order_by_current_frame_else_frame_0)[-1]
    else:
        instance = instances[0]

    # Read the actual string caption
    caption = instance.get_answer()

    # Send the first frame of the video along with the human caption to the LLM.
    frame = Frame(frame=0, content=frame_content)
    response = prompt_gpt(caption, frame)

    # Process the LLM's response, which contains three different rephrasings of the original caption.
    # Update the label row with the new captions, replacing any existing ones.
    for r, t in zip(rs, [response.rephrase_1, response.rephrase_2, response.rephrase_3]):
        # Overwrite any existing re-captions
        existing_instances = label_row.get_classification_instances(filter_ontology_classification=r)
        for existing_instance in existing_instances:
            label_row.remove_classification(existing_instance)

        # Create new instances
        ins = r.create_instance()
        ins.set_answer(t, attribute=r.attributes[0])
        ins.set_for_frames(0)
        label_row.add_classification_instance(ins)

    label_row.save()

ENCORD_SSH_KEY_FILE=/path/to/your_private_key \
OPENAI_API_KEY=<your-api-key> \
fastapi dev main.py

source venv/bin/activate
encord-agents test local my_agent '<your_url>'

python -m venv venv
source venv/bin/activate
python -m pip install encord-agents modal

git clone https://github.com/facebookresearch/co-tracker.git
mv co-tracker/cotracker ./cotracker

from pathlib import Path

import modal
from encord.objects.coordinates import (
    PointCoordinate,
)
from encord.objects.ontology_labels_impl import LabelRowV2
from encord.objects.ontology_object_instance import ObjectInstance
from fastapi import Depends
from typing_extensions import Annotated

from encord_agents.fastapi.dependencies import (
    FrameData,
    dep_asset,
    dep_label_row,
    dep_objects,
)

# 1. Define the Modal image.
# This specifies the base environment, installs system dependencies,
# downloads the CoTracker3 model weights, and installs Python packages.
image = (
    modal.Image.debian_slim(python_version="3.12")
    .apt_install("libgl1", "libglib2.0-0", "wget")
    .run_commands(
        "wget https://huggingface.co/facebook/cotracker3/resolve/main/scaled_offline.pth",
    )
    .pip_install(
        "fastapi[standard]",
        "encord-agents",
        "torch",
        "torchvision",
        "tqdm",
        "imageio[ffmpeg]",
    )
    .add_local_python_source("cotracker") # Assuming 'cotracker' is a local directory with CoTracker source
)

# 2. Define the Modal app.
# This creates the Modal application instance, linking it to the defined image.
app = modal.App(name="encord-agents-cotracker-3-with-model", image=image)


# Helper function to read video frames from a given path using imageio.
def read_video_from_path(path):
    import imageio
    import numpy as np

    try:
        reader = imageio.get_reader(path)
    except Exception as e:
        print("Error opening video file: ", e)
        return None
    frames = []
    for i, im in enumerate(reader):
        frames.append(np.array(im))
    return np.stack(frames)


# 3. Define the endpoint and CoTracker3 usage.
# This is the main function that runs on Modal, exposed as a web endpoint.
@app.function(
    secrets=[modal.Secret.from_name("encord-ssh-key")], # Accesses a Modal secret for Encord authentication
    gpu="L4" # Specifies that the function requires a GPU (L4 type)
)
@modal.web_endpoint(method="POST")
def cotracker3(
    frame_data: FrameData, # Injected frame metadata from Encord webhook
    lr: Annotated[LabelRowV2, Depends(dep_label_row)], # Injected label row object
    object_instances: Annotated[list[ObjectInstance], Depends(dep_objects)], # Injected list of selected objects
    asset: Annotated[Path, Depends(dep_asset)], # Injected path to the local video asset
):
    import imageio
    import numpy
    import torch
    from cotracker.predictor import CoTrackerPredictor

    # Initialize CoTracker3 model, moving to GPU if available
    model = CoTrackerPredictor(checkpoint="/scaled_offline.pth")
    if torch.cuda.is_available():
        model = model.cuda()

    # Ensure only one object is selected for tracking
    assert len(object_instances) == 1
    obj_inst = object_instances[0]
    
    # Read the video from the asset path and convert to PyTorch tensor
    video = read_video_from_path(asset)
    video_tensor = torch.from_numpy(video).permute(0, 3, 1, 2)[None].float()

    # Move video tensor to GPU if available
    if torch.cuda.is_available():
        video_tensor = video_tensor.cuda()
        
    # Extract query point from the selected object instance's annotation
    annotation = obj_inst.get_annotation(frame_data.frame)
    assert isinstance(annotation.coordinates, PointCoordinate)
    assert lr.width # Ensure label row dimensions are available
    assert lr.height

    # Prepare the query tensor for CoTracker (frame number, x-coordinate, y-coordinate)
    query = torch.tensor(
        [
            [
                frame_data.frame,
                annotation.coordinates.x * lr.width, # Convert normalized x to pixel x
                annotation.coordinates.y * lr.height, # Convert normalized y to pixel y
            ],
        ]
    )
    if torch.cuda.is_available():
        query = query.cuda()
        
    # Run CoTracker to predict tracks based on the query point
    pred_tracks, _ = model(video_tensor, queries=query[None])
    
    # Update the object instance with predicted tracks for each frame
    for frame_num, coord in enumerate(pred_tracks.reshape(-1, 2)):
        try:
            obj_inst.set_for_frames(
                coordinates=PointCoordinate(x=float(coord[0]) / lr.width, y=float(coord[1]) / lr.height),
                frames=frame_num,
            )
        except Exception:
            # Skip frames where updating might fail (e.g., if coordinates are out of bounds)
            continue
            
    # Save the updated label row with the new tracked object instance
    lr.save()