Implementation:Microsoft Autogen EvalOrchestrator

Knowledge Sources	Microsoft_Autogen
Domains	Evaluation, Orchestration, Task Management, Async Operations
Last Updated	2026-02-11 17:00 GMT

Overview

EvalOrchestrator manages the lifecycle of evaluation tasks, criteria, and runs for AutoGen Studio's evaluation framework with support for both database persistence and in-memory storage.

Description

The EvalOrchestrator class serves as the central coordinator for the evaluation system in AutoGen Studio. It manages evaluation tasks (test scenarios), evaluation criteria (judging dimensions), and evaluation runs (execution instances). The orchestrator supports both persistent storage via DatabaseManager and in-memory operation when no database is available. It handles asynchronous execution of evaluation runs, tracks active runs, coordinates between runners (which execute tasks) and judges (which score results), and provides methods to tabulate results across multiple runs for visualization and comparison. The orchestrator maintains the complete evaluation workflow from task creation through execution to scoring and result storage.

Usage

Use EvalOrchestrator when building evaluation pipelines, managing agent performance testing, coordinating evaluation runs across multiple tasks and criteria, tracking evaluation execution status, or generating comparative analysis of evaluation results.

Code Reference

Source Location

• Repository: Microsoft_Autogen
• File: python/packages/autogen-studio/autogenstudio/eval/orchestrator.py
• Lines: 1-789

Signature

class EvalOrchestrator:
    def __init__(self, db_manager: Optional[DatabaseManager] = None)

    # Task Management
    async def create_task(self, task: EvalTask) -> str
    async def get_task(self, task_id: str) -> Optional[EvalTask]
    async def list_tasks(self) -> List[EvalTask]

    # Criteria Management
    async def create_criteria(self, criteria: EvalJudgeCriteria) -> str
    async def get_criteria(self, criteria_id: str) -> Optional[EvalJudgeCriteria]
    async def list_criteria(self) -> List[EvalJudgeCriteria]

    # Run Management
    async def create_run(
        self,
        task: Union[str, EvalTask],
        runner: BaseEvalRunner,
        judge: BaseEvalJudge,
        criteria: List[Union[str, EvalJudgeCriteria]],
        name: str = "",
        description: str = "",
    ) -> str

    async def start_run(self, run_id: str) -> None
    async def get_run_status(self, run_id: str) -> Optional[EvalRunStatus]
    async def get_run_result(self, run_id: str) -> Optional[EvalRunResult]
    async def get_run_score(self, run_id: str) -> Optional[EvalScore]
    async def list_runs(self) -> List[Dict[str, Any]]
    async def cancel_run(self, run_id: str) -> bool

    # Results
    async def tabulate_results(
        self,
        run_ids: List[str],
        include_reasons: bool = False
    ) -> TabulatedResults

Import

from autogenstudio.eval.orchestrator import EvalOrchestrator
from autogenstudio.database import DatabaseManager

# With database persistence
db_manager = DatabaseManager(engine_uri="sqlite:///./eval.db")
orchestrator = EvalOrchestrator(db_manager=db_manager)

# In-memory only
orchestrator = EvalOrchestrator()

I/O Contract

Inputs

Name	Type	Required	Description
db_manager	Optional[DatabaseManager]	No	Database manager for persistence; if None, data stored in memory only
task	Union[str, EvalTask]	Yes	Task to evaluate (ID or task object)
task_id	str	Yes	ID of the task to retrieve
criteria	Union[str, EvalJudgeCriteria]	Yes	Evaluation criteria (ID or criteria object)
criteria_id	str	Yes	ID of the criteria to retrieve
runner	BaseEvalRunner	Yes	Runner to execute evaluation
judge	BaseEvalJudge	Yes	Judge to score evaluation results
run_id	str	Yes	ID of the run to query or control
run_ids	List[str]	Yes	List of run IDs for tabulation
include_reasons	bool	No	Whether to include scoring reasons in tabulated results
name	str	No	Name for the run
description	str	No	Description for the run

Outputs

Name	Type	Description
create_task	str	Task ID for the created task
get_task	Optional[EvalTask]	Task object if found, None otherwise
list_tasks	List[EvalTask]	List of all evaluation tasks
create_criteria	str	Criteria ID for the created criteria
get_criteria	Optional[EvalJudgeCriteria]	Criteria object if found, None otherwise
list_criteria	List[EvalJudgeCriteria]	List of all evaluation criteria
create_run	str	Run ID for the created run
get_run_status	Optional[EvalRunStatus]	Run status if found, None otherwise
get_run_result	Optional[EvalRunResult]	Run execution result if found, None otherwise
get_run_score	Optional[EvalScore]	Run evaluation score if found, None otherwise
list_runs	List[Dict[str, Any]]	List of run configurations with metadata
cancel_run	bool	True if run was cancelled, False otherwise
tabulate_results	TabulatedResults	Structured data with dimensions and run scores for visualization

Usage Examples

Creating and Running Evaluations

import asyncio
from autogenstudio.eval.orchestrator import EvalOrchestrator
from autogenstudio.datamodel.eval import EvalTask, EvalJudgeCriteria
from autogenstudio.eval.runners import ModelEvalRunner
from autogenstudio.eval.judges import LLMEvalJudge

# Initialize orchestrator
orchestrator = EvalOrchestrator()

async def run_evaluation():
    # Create evaluation task
    task = EvalTask(
        name="Customer Support Test",
        description="Test agent's ability to handle customer inquiries",
        config={
            "prompt": "Help a customer who is upset about a delayed order",
            "expected_behaviors": ["empathy", "problem_solving"]
        }
    )
    task_id = await orchestrator.create_task(task)

    # Create evaluation criteria
    criteria = EvalJudgeCriteria(
        dimension="Empathy",
        prompt="Rate the agent's empathy in the response (0-10)",
        scale_description="0=No empathy, 10=Exceptional empathy"
    )
    criteria_id = await orchestrator.create_criteria(criteria)

    # Create runner and judge
    runner = ModelEvalRunner()
    judge = LLMEvalJudge()

    # Create and start run
    run_id = await orchestrator.create_run(
        task=task_id,
        runner=runner,
        judge=judge,
        criteria=[criteria_id],
        name="Test Run 1"
    )

    await orchestrator.start_run(run_id)

    # Monitor status
    while True:
        status = await orchestrator.get_run_status(run_id)
        print(f"Run status: {status}")
        if status in ["COMPLETED", "FAILED", "CANCELED"]:
            break
        await asyncio.sleep(1)

    # Get results
    result = await orchestrator.get_run_result(run_id)
    score = await orchestrator.get_run_score(run_id)

    print(f"Result: {result}")
    print(f"Score: {score}")

asyncio.run(run_evaluation())

Managing Multiple Evaluation Runs

async def compare_models():
    orchestrator = EvalOrchestrator()

    # Create a task
    task = EvalTask(name="Code Generation", description="Generate Python functions")
    task_id = await orchestrator.create_task(task)

    # Create criteria
    quality_criteria = EvalJudgeCriteria(
        dimension="Code Quality",
        prompt="Rate code quality (0-10)"
    )
    quality_id = await orchestrator.create_criteria(quality_criteria)

    # Create runs for different models
    run_ids = []
    for model_name in ["gpt-4", "claude-3", "gemini-pro"]:
        runner = ModelEvalRunner(model=model_name)
        judge = LLMEvalJudge()

        run_id = await orchestrator.create_run(
            task=task_id,
            runner=runner,
            judge=judge,
            criteria=[quality_id],
            name=f"{model_name} Run"
        )
        run_ids.append(run_id)
        await orchestrator.start_run(run_id)

    # Wait for all runs to complete
    # (in production, use proper async monitoring)
    await asyncio.sleep(10)

    # Tabulate results
    results = await orchestrator.tabulate_results(
        run_ids=run_ids,
        include_reasons=True
    )

    print("Dimensions:", results["dimensions"])
    for run in results["runs"]:
        print(f"\n{run['name']}:")
        print(f"  Overall: {run['overall_score']}")
        print(f"  Scores: {run['scores']}")

asyncio.run(compare_models())

Listing and Managing Evaluation Assets

async def manage_evaluation_assets():
    orchestrator = EvalOrchestrator()

    # List all tasks
    tasks = await orchestrator.list_tasks()
    print(f"Available tasks: {len(tasks)}")
    for task in tasks:
        print(f"  - {task.name}: {task.description}")

    # List all criteria
    criteria_list = await orchestrator.list_criteria()
    print(f"\nAvailable criteria: {len(criteria_list)}")
    for criteria in criteria_list:
        print(f"  - {criteria.dimension}")

    # List all runs
    runs = await orchestrator.list_runs()
    print(f"\nAvailable runs: {len(runs)}")
    for run in runs:
        print(f"  - {run['name']} (Status: {run['status']})")

    # Cancel an active run if needed
    active_runs = [r for r in runs if r['status'] == 'RUNNING']
    if active_runs:
        run_to_cancel = active_runs[0]['id']
        cancelled = await orchestrator.cancel_run(run_to_cancel)
        print(f"\nCancelled run {run_to_cancel}: {cancelled}")

asyncio.run(manage_evaluation_assets())

Tabulating Results for Visualization

async def generate_comparison_data():
    orchestrator = EvalOrchestrator()

    # Get run IDs from previous evaluations
    all_runs = await orchestrator.list_runs()
    completed_runs = [r['id'] for r in all_runs if r['status'] == 'COMPLETED']

    # Tabulate results
    tabulated = await orchestrator.tabulate_results(
        run_ids=completed_runs[:5],  # Compare first 5 runs
        include_reasons=True
    )

    # Format for radar chart or comparison table
    print("Evaluation Dimensions:", tabulated["dimensions"])
    print("\nRun Comparison:")

    for run in tabulated["runs"]:
        print(f"\n{run['name']} ({run['runner_type']}):")
        print(f"  Task: {run['task_name']}")
        print(f"  Overall Score: {run['overall_score']}")

        for dim, score, reason in zip(
            tabulated["dimensions"],
            run["scores"],
            run["reasons"]
        ):
            print(f"  {dim}: {score}")
            if reason:
                print(f"    Reason: {reason}")

asyncio.run(generate_comparison_data())

Related Pages

• BaseEvalRunner - Base class for evaluation runners
• BaseEvalJudge - Base class for evaluation judges
• EvalTask - Task data model for evaluations
• EvalJudgeCriteria - Criteria data model for evaluations
• DatabaseManager - Provides database persistence