Adala: Autonomous Data Labeling Agent

The Worker Pool is a scalable architecture designed to process AI inference tasks from Label Studio Enterprise (LSE). It uses a 'Competing Consumers' pattern to achieve horizontal scalability and natural load balancing.

Architecture Flow

1. Submission: LSE uses the AdalaClient to call the Adala API (POST /worker-pool/submit-batch).
2. Distribution: The API publishes work messages to a shared Kafka input topic (worker_pool_input).
3. Processing: Multiple Celery workers compete for messages from the input topic. Each worker runs a WorkerProcessor that performs AI inference (e.g., using OpenAI or Anthropic).
4. Result Handling: Workers pass results through an internal async queue to an OutputProcessor.
5. Completion: The OutputProcessor groups results by modelrun_id and sends them back to LSE via its API (/api/prompts/{modelrun_id}/batch-predictions).

Key Components

• AdalaClient: The interface located in label_studio_enterprise.lse_ml_models.adala_client used to prepare payloads and handle errors.
• Celery Workers: Forever-running processes that host both the WorkerProcessor and OutputProcessor via celery_integration.py.
• Kafka Topics: worker_pool_input (work distribution) and worker_pool_output (results).

You can use Adala directly within a Python environment (like a Jupyter notebook) by importing its core components: Agent, StaticEnvironment, ClassificationSkill, and OpenAIChatRuntime.

To set up an agent, you need to:

1. Prepare Datasets: Create pandas.DataFrame objects for training (ground truth) and testing.
2. Initialize an Agent: Pass a StaticEnvironment containing your training data, a list of skills, and a dictionary of runtimes (e.g., OpenAIChatRuntime).
3. Define Skills: Use specialized skill classes like ClassificationSkill to define instructions, labels, and input/output templates.
4. Train and Run: Call agent.learn() to optimize the skill based on the training data, then use agent.run() to generate predictions on the test dataset.

import pandas as pd
from adala.agents import Agent
from adala.environments import StaticEnvironment
from adala.skills import ClassificationSkill
from adala.runtimes import OpenAIChatRuntime
from rich import print

# Train dataset
train_df = pd.DataFrame([
    ["It was the negative first impressions, and then it started working.", "Positive"],
    ["Not loud enough and doesn't turn on like it should.", "Negative"],
    ["I don't know what to say.", "Neutral"],
    ["Manager was rude, but the most important that mic shows very flat frequency response.", "Positive"],
    ["The phone doesn't seem to accept anything except CBR mp3s.", "Negative"],
    ["I tried it before, I bought this device for my son.", "Neutral"],
], columns=["text", "sentiment"])

# Test dataset
test_df = pd.DataFrame([
    "All three broke within two months of use.",
    "The device worked for a long time, can't say anything bad.",
    "Just a random line of text."
], columns=["text"])

agent = Agent(
    environment=StaticEnvironment(df=train_df),
    skills=ClassificationSkill(
        name='sentiment',
        instructions="Label text as positive, negative or neutral.",
        labels=["Positive", "Negative", "Neutral"],
        input_template="Text: {text}",
        output_template="Sentiment: {sentiment}"
    ),
    runtimes = {
        'openai': OpenAIChatRuntime(model='gpt-4o'),
    },
    teacher_runtimes = {
        'default': OpenAIChatRuntime(model='gpt-4o'),
    },
    default_runtime='openai',
)

agent.learn(learning_iterations=3, accuracy_threshold=0.95)
predictions = agent.run(test_df)
print(predictions)

To use Claude, Gemini, or other OpenAI-compatible models via OpenRouter, set the OPENROUTER_API_KEY environment variable and configure OpenAIChatRuntime with the OpenRouter base_url and the specific model name.

1. Set Environment Variable:

   export OPENROUTER_API_KEY='your-openrouter-api-key'

2. Configure Runtime: In your Agent definition, set the base_url to https://openrouter.ai/api/v1 and set the provider to "Custom".

import os
import pandas as pd
from adala.agents import Agent
from adala.environments import StaticEnvironment
from adala.skills import ClassificationSkill
from adala.runtimes import OpenAIChatRuntime

# ... (train_df and test_df definitions) ...

agent = Agent(
    environment=StaticEnvironment(df=train_df),
    skills=ClassificationSkill(
        name='sentiment',
        instructions="Label text as positive, negative or neutral.",
        labels=["Positive", "Negative", "Neutral"],
        input_template="Text: {text}",
        output_template="Sentiment: {sentiment}"
    ),
    runtimes = {
        'openrouter': OpenAIChatRuntime(
            base_url="https://openrouter.ai/api/v1",
            model="anthropic/claude-3.5-haiku",
            api_key=os.getenv("OPENROUTER_API_KEY"),
            provider="Custom"
        ),
    },
    default_runtime='openrouter',
    teacher_runtimes = {
        "default" : OpenAIChatRuntime(
            base_url="https://openrouter.ai/api/v1",
            model="anthropic/claude-3.5-haiku",
            api_key=os.getenv("OPENROUTER_API_KEY"),
            provider="Custom"
        ),
    }
)

agent.learn(learning_iterations=3, accuracy_threshold=0.95)
predictions = agent.run(test_df)

When training an agent to learn a sequence of skills, use StaticEnvironment to provide ground truth demonstrations. This allows the agent to compare its generated outputs against expected values.

Key parameters for StaticEnvironment:

• df: A pandas.DataFrame containing the input data and the expected outputs for each skill.
• ground_truth_columns: A dictionary mapping the skill names (e.g., skill_0) to the corresponding column names in the DataFrame that contain the correct answers.
• matching_function: The method used to compare agent output to ground truth (e.g., 'fuzzy').
• matching_threshold: A float representing the required similarity score (e.g., 0.9) for a match to be considered successful.

from adala.environments import StaticEnvironment
import pandas as pd

environment = StaticEnvironment(
    df=pd.DataFrame([
        {
          "category": "Macronutrients",
          "entities": "Carbohydrates, Proteins, Fats",
          "text": "Carbohydrates provide quick energy..."
        }
    ]),
    ground_truth_columns={
        'skill_0': 'entities',
        'skill_1': 'text'
    },
    matching_function='fuzzy',
    matching_threshold=0.9
)