samelhousseini avatar

workflow-metrics

Track, evaluate, and analyze multi-agent AI workflows using Azure AI Foundry cloud evaluations. Use

by samelhousseini|Open Source

Workflow Metrics Skill

Folder Contents

FileTypeDescription
SKILL.mdDocumentationMain skill documentation with 14 evaluation metrics, trajectory format, and decorator usage
PRD.mdDocumentationProduct Requirements Document for the skill
.env.sampleConfigurationSample environment variables (AZURE_AI_PROJECT_ENDPOINT)
requirements.txtDependenciesPython package dependencies (azure-ai-projects, azure-identity)
scripts/workflow_metrics/
scripts/workflow_metrics/__init__.pyModulePackage initializer with exports for all classes and decorators
scripts/workflow_metrics/trajectory.pyLoggingWorkflowTrajectoryLog class for multi-step agent trajectory logging in Foundry-compatible format
scripts/workflow_metrics/metrics.pyMetricsTrajectoryMetrics class for calculating metrics from trajectory files
scripts/workflow_metrics/foundry_evals.pyEvaluatorFoundryEvalsClient for Azure AI Foundry cloud evaluation API with 14 evaluation metrics
scripts/workflow_metrics/telemetry.pyDecorators@track_workflow decorator and context functions (log_plan, log_step, auto_log_step)
scripts/workflow_metrics/trajectory_manager.pyManagerEvaluationManager singleton for daily metrics files management
scripts/workflow_metrics/tool_definitions_helper.pyHelperGenerate tool definitions from Python functions with optional LLM enrichment
scripts/workflow_metrics/utils.pyUtilitiesUtility functions for workflow execution and common operations

Track, evaluate, and analyze multi-agent AI workflows using Azure AI Foundry cloud evaluations.

Table of Contents

  1. Overview
  2. Module Location
  3. Quick Start: Cloud Eval Runs
  4. Available Cloud Metrics
  5. FoundryEvalsClient: Complete API Reference
  6. Creating Cloud Eval Runs Step-by-Step
  7. Data Format for Cloud Evaluations
  8. Metric Configurations
  9. Result Parsing
  10. Trajectory Logging (for Cloud Evaluation)
  11. Environment Variables
  12. Local Metrics (Only If Specifically Requested)
  13. Files Reference
  14. Related Resources

Overview

This skill provides cloud-based evaluation of AI agent workflows via Azure AI Foundry Evals API.

CRITICAL: All evaluations (13 LLM-as-judge metrics) MUST be run in Azure AI Foundry cloud. Local SDK-based evaluations are NOT supported. Always use FoundryEvalsClient to create cloud eval runs.

The evaluation flow:

  1. Log trajectories in Foundry-compatible format
  2. Create an eval group with testing criteria (metrics)
  3. Submit an eval run with inline JSONL data
  4. Poll for completion and retrieve results

Module Location

.github/skills/workflow-metrics/scripts/workflow_metrics/

Quick Start: Cloud Eval Runs

from workflow_metrics import FoundryEvalsClient, WorkflowTrajectoryLog

# 1. Log a trajectory
trajectory = WorkflowTrajectoryLog(
    user_query="What's the status of order W123456?",
    system_prompt="You are a helpful customer service agent."
)
trajectory.log_plan("Check order status", ["get_order_status"])
trajectory.log_step(tool_name="get_order_status", arguments={"order_id": "W123456"}, result={"status": "shipped"})
trajectory.finalize("Your order W123456 has been shipped!")

# 2. Create cloud eval run
client = FoundryEvalsClient()

# Create eval group with metrics
eval_id = client.create_eval(
    name="order_workflow_eval",
    metrics=["coherence", "task_completion", "tool_call_accuracy"]
)

# Submit the workflow for cloud evaluation
workflow_data = trajectory.to_dict()
run_id = client.run_eval(
    eval_id=eval_id,
    items=[client._prepare_workflow_data(workflow_data)],
    run_name="order_eval_run"
)

# Wait for cloud completion and get results
results = client.wait_for_completion(eval_id=eval_id, run_id=run_id)
print(results)

Available Cloud Metrics

All 13 metrics below run in Azure AI Foundry cloud as LLM-as-judge evaluations:

Text Quality Metrics

MetricEvaluatorDescriptionRequired Fields
coherencebuiltin.coherenceLogical flow and consistencyquery, response
fluencybuiltin.fluencyGrammar and readabilityresponse
relevancebuiltin.relevanceResponse relevance to queryquery, response
groundednessbuiltin.groundednessFactual accuracy based on contextresponse (context, query, tool_definitions optional)

Agent Quality Metrics

MetricEvaluatorDescriptionRequired Fields
intent_resolutionbuiltin.intent_resolutionHow well agent understood intentquery, response
task_adherencebuiltin.task_adherenceFollowing task requirementsquery, response
task_completionbuiltin.task_completionSuccessfully completing tasksquery, response
response_completenessbuiltin.response_completenessCompleteness of responseground_truth, response

Tool Usage Metrics

MetricEvaluatorDescriptionRequired Fields
tool_call_accuracybuiltin.tool_call_accuracyCorrect tool selectionquery, tool_definitions
tool_input_accuracybuiltin.tool_input_accuracyCorrect tool inputsquery, response, tool_definitions
tool_output_utilizationbuiltin.tool_output_utilizationEffective use of tool resultsquery, response
tool_selectionbuiltin.tool_selectionAppropriate tool choicesquery, response, tool_definitions
tool_successbuiltin.tool_call_successTool execution success rateresponse

FoundryEvalsClient: Complete API Reference

The FoundryEvalsClient class in foundry_evals.py is the primary interface for cloud evaluations.

Initialization

from workflow_metrics import FoundryEvalsClient

client = FoundryEvalsClient(
    endpoint=None,              # Uses AZURE_AI_PROJECT_ENDPOINT env var
    deployment_name=None,       # Uses AZURE_AI_MODEL_DEPLOYMENT_NAME env var (default: "gpt-4.1")
    use_reasoning_model=False,  # Set True for o-series models
    threshold=3.0,              # Score threshold for pass/fail (1-5 scale)
    matching_mode="in_order_match"  # For task_navigation_efficiency
)

Core Methods

create_eval(name, metrics) -> str

Creates an evaluation group with specified metrics (testing criteria).

eval_id = client.create_eval(
    name="my_agent_eval",
    metrics=["coherence", "task_completion", "tool_call_accuracy"]
)
# Returns: "eval_abc123..."

What happens internally:

# Builds data_source_config schema based on required fields
data_source_config = {
    "type": "custom",
    "item_schema": {
        "type": "object",
        "properties": {...},  # Fields needed by selected metrics
        "required": [...]
    },
    "include_sample_schema": True
}

# Builds testing_criteria for each metric
testing_criteria = [
    {
        "type": "azure_ai_evaluator",
        "name": "coherence",
        "evaluator_name": "builtin.coherence",
        "initialization_parameters": {
            "deployment_name": "gpt-4.1"
        },
        "data_mapping": {
            "query": "{{item.query}}",
            "response": "{{item.response}}"
        }
    },
    # ... more criteria
]

eval_obj = client.evals.create(
    name=name,
    data_source_config=data_source_config,
    testing_criteria=testing_criteria
)

run_eval(eval_id, items, run_name, metadata) -> str

Submits items for cloud evaluation using inline JSONL data.

run_id = client.run_eval(
    eval_id=eval_id,
    items=prepared_items,  # List of workflow data dicts
    run_name="my_run_001",
    metadata={"source": "production", "version": "1.0"}
)
# Returns: "run_xyz789..."

Critical: The API call structure:

from openai.types.evals.create_eval_jsonl_run_data_source_param import (
    CreateEvalJSONLRunDataSourceParam,
    SourceFileContent,
    SourceFileContentContent,
)

# Build content items (filter out internal tracking fields)
content_items = []
for item in items:
    eval_item = {k: v for k, v in item.items() if not k.startswith("_")}
    content_items.append(SourceFileContentContent(item=eval_item))

# Create the eval run with inline JSONL data
eval_run = client.evals.runs.create(
    eval_id=eval_id,
    name=run_name or f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
    metadata=metadata or {"source": "FoundryEvalsClient"},
    data_source=CreateEvalJSONLRunDataSourceParam(
        type="jsonl",
        source=SourceFileContent(
            type="file_content",      # Inline content, not file reference
            content=content_items,    # List of SourceFileContentContent
        ),
    ),
)

wait_for_completion(eval_id, run_id, timeout, poll_interval) -> Dict

Polls until evaluation completes and retrieves results.

result = client.wait_for_completion(
    eval_id=eval_id,
    run_id=run_id,
    timeout=300,        # Max wait in seconds
    poll_interval=5     # Seconds between polls
)
# Returns: {
#     "status": "completed" | "failed" | "timeout",
#     "output_items": [...],
#     "report_url": "https://...",
#     "eval_id": "...",
#     "run_id": "..."
# }

evaluate_workflows(workflows, metrics, batch_size, timeout) -> Dict

High-level method that handles the full evaluation flow.

results = client.evaluate_workflows(
    workflows=workflow_list,    # List of workflow dicts with foundry_format
    metrics=["coherence", "task_completion"],
    batch_size=50,              # Max workflows per eval run
    timeout=300                 # Max wait per batch
)
# Returns: {
#     "workflow_run_id_1": {
#         "coherence": {"score": 4.0, "pass": True, "reason": "..."},
#         "task_completion": {"score": 5.0, "pass": True, "reason": "..."}
#     },
#     ...
# }

evaluate_single(workflow, metrics, timeout) -> Dict

Evaluate a single workflow.

result = client.evaluate_single(
    workflow=workflow_dict,
    metrics=["coherence", "tool_call_accuracy"],
    timeout=60
)

Creating Cloud Eval Runs Step-by-Step

Step 1: Prepare Workflow Data

Workflows must have a foundry_format section:

workflow = {
    "workflow_id": "wf_001",
    "workflow_run_id": "run_001",
    "user_request": "Check order status",
    "foundry_format": {
        "query": "What is the status of order W123?",
        "response": [
            {"role": "assistant", "content": "Let me check that for you."},
            {"role": "assistant", "content": [
                {
                    "type": "tool_call",
                    "tool_call_id": "tc_1",
                    "name": "get_order_status",
                    "arguments": {"order_id": "W123"}
                }
            ]},
            {"role": "tool", "content": [
                {
                    "type": "tool_result",
                    "tool_call_id": "tc_1",
                    "tool_result": {"status": "shipped", "tracking": "1Z999"}
                }
            ]},
            {"role": "assistant", "content": "Your order W123 has been shipped! Tracking: 1Z999"}
        ],
        "tool_definitions": [
            {
                "name": "get_order_status",
                "description": "Get the status of an order",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {"type": "string", "description": "Order ID"}
                    },
                    "required": ["order_id"]
                }
            }
        ],
        "ground_truth": ["get_order_status"]  # Expected tool sequence
    }
}

Step 2: Create Eval Group

client = FoundryEvalsClient()

# Select metrics to evaluate
metrics = [
    "coherence",           # Text quality
    "task_completion",     # Did agent complete the task?
    "tool_call_accuracy",  # Did agent call the right tools?
    "tool_input_accuracy"  # Were tool inputs correct?
]

eval_id = client.create_eval(
    name=f"workflow_eval_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
    metrics=metrics
)
print(f"Created eval group: {eval_id}")

Step 3: Submit Eval Run

# Prepare items for the API
prepared_items = [client._prepare_workflow_data(w) for w in workflows]

# Submit to Azure AI Foundry cloud
run_id = client.run_eval(
    eval_id=eval_id,
    items=prepared_items,
    run_name="batch_1",
    metadata={"environment": "production", "version": "2.0"}
)
print(f"Submitted eval run: {run_id}")

Step 4: Wait for Results

# Poll until completion (runs in Azure cloud)
completion = client.wait_for_completion(
    eval_id=eval_id,
    run_id=run_id,
    timeout=300,
    poll_interval=5
)

if completion["status"] == "completed":
    print(f"Eval completed! Report: {completion['report_url']}")

    # Parse results
    results = client._parse_results(
        completion["output_items"],
        prepared_items,
        metrics
    )

    for workflow_id, metric_results in results.items():
        print(f"\nWorkflow: {workflow_id}")
        for metric, result in metric_results.items():
            print(f"  {metric}: {result['score']:.1f} ({'PASS' if result['pass'] else 'FAIL'})")
else:
    print(f"Eval {completion['status']}: {completion.get('error', '')}")

Data Format for Cloud Evaluations

Required Fields by Metric

Metricqueryresponsetool_definitionsground_truthcontext
coherence✓✓
fluency✓
relevance✓✓
groundedness✓(opt)(opt)
intent_resolution✓✓(opt)
task_adherence✓✓(opt)
task_completion✓✓(opt)
response_completeness✓✓
tool_call_accuracy✓(opt)✓
tool_input_accuracy✓✓✓
tool_output_utilization✓✓(opt)
tool_selection✓✓✓
tool_success✓(opt)

Response Format (Conversation Array)

The response field must be an array of message objects:

response = [
    # Assistant text message
    {"role": "assistant", "content": "I'll help you with that."},

    # Assistant tool call
    {"role": "assistant", "content": [
        {
            "type": "tool_call",
            "tool_call_id": "call_001",
            "name": "search_orders",
            "arguments": {"customer_id": "C123"}
        }
    ]},

    # Tool result
    {"role": "tool", "content": [
        {
            "type": "tool_result",
            "tool_call_id": "call_001",
            "tool_result": {"orders": [{"id": "W123", "status": "shipped"}]}
        }
    ]},

    # Final assistant response
    {"role": "assistant", "content": "I found your order W123. It has been shipped!"}
]

Tool Definitions Format

tool_definitions = [
    {
        "name": "search_orders",
        "description": "Search for customer orders",
        "parameters": {
            "type": "object",
            "properties": {
                "customer_id": {
                    "type": "string",
                    "description": "Customer ID to search"
                },
                "status": {
                    "type": "string",
                    "enum": ["pending", "shipped", "delivered"],
                    "description": "Filter by status"
                }
            },
            "required": ["customer_id"]
        }
    }
]

Metric Configurations

The FOUNDRY_METRICS dictionary in foundry_evals.py defines each metric's configuration:

FOUNDRY_METRICS = {
    "coherence": {
        "evaluator_name": "builtin.coherence",
        "data_mapping": {
            "query": "{{item.query}}",
            "response": "{{item.response}}",
        },
        "requires": ["query", "response"],
    },
    "tool_call_accuracy": {
        "evaluator_name": "builtin.tool_call_accuracy",
        "data_mapping": {
            "query": "{{item.query}}",
            "tool_definitions": "{{item.tool_definitions}}",
            "tool_calls": "{{item.tool_calls}}",
            "response": "{{item.response}}",
        },
        "requires": ["query", "tool_definitions"],
    },
    # ... 11 more metrics
}

Testing Criteria Structure

Each metric becomes a testing criterion:

{
    "type": "azure_ai_evaluator",
    "name": "tool_call_accuracy",
    "evaluator_name": "builtin.tool_call_accuracy",
    "initialization_parameters": {
        "deployment_name": "gpt-4.1",
        # "is_reasoning_model": True  # For o-series models
    },
    "data_mapping": {
        "query": "{{item.query}}",
        "tool_definitions": "{{item.tool_definitions}}",
        "tool_calls": "{{item.tool_calls}}",
        "response": "{{item.response}}"
    }
}

Result Parsing

Azure AI Foundry returns different result formats depending on the metric:

Score Format (1-5)

Used by: coherence, fluency, relevance, groundedness, intent_resolution, response_completeness, tool_selection

{"score": 4, "explanation": "The response is coherent and well-structured..."}

Flagged Format

Used by: task_adherence

{"flagged": False, "reasoning": "The agent followed all task requirements..."}
# flagged=False means PASS (score=5), flagged=True means FAIL (score=1)

Success Boolean Format

Used by: task_completion, tool_success

{"success": True, "explanation": "The task was completed successfully..."}
# success=True -> score=5, success=False -> score=1

Tool Level Format

Used by: tool_call_accuracy

{"tool_calls_success_level": 4, "chain_of_thought": "The agent selected appropriate tools..."}

Pass/Fail String Format

Used by: tool_input_accuracy

{"result": "pass", "chain_of_thought": "All tool inputs were accurate..."}

Used by: tool_output_utilization

{"label": "pass", "reason": "The agent effectively used tool outputs..."}

Trajectory Logging (for Cloud Evaluation)

Use WorkflowTrajectoryLog to create Foundry-compatible trajectory data:

from workflow_metrics import WorkflowTrajectoryLog

trajectory = WorkflowTrajectoryLog(
    user_query="Cancel order W123",
    system_prompt="You are a customer service agent.",
    tool_definitions=[{
        "name": "cancel_order",
        "description": "Cancel a customer order",
        "parameters": {...}
    }]
)

# Log plan
trajectory.log_plan("Cancel the order", ["cancel_order"])

# Log tool execution
trajectory.log_step(
    tool_name="cancel_order",
    arguments={"order_id": "W123", "reason": "Customer request"},
    result={"success": True, "refund_amount": 99.99},
    success=True
)

# Finalize
trajectory.finalize("Order W123 has been cancelled. Refund of $99.99 will be processed.")

# Get dict for cloud evaluation
workflow_dict = trajectory.to_dict()

# Or save to file
trajectory.save("trajectories/cancellation.jsonl")

Decorator-Based Logging

from workflow_metrics import track_workflow, log_plan, auto_log_step

@track_workflow(output_dir="trajectories")
async def my_workflow(user_request: str):
    log_plan(user_request, ["tool_a", "tool_b"])

    result = await tool_a()
    auto_log_step(result)

    result = await tool_b()
    auto_log_step(result)

    return "Done"

Environment Variables

# REQUIRED for Azure AI Foundry cloud evaluations
AZURE_AI_PROJECT_ENDPOINT=https://your-project.api.azureml.ms

# Optional: Model deployment for LLM-as-judge (default: gpt-4.1)
AZURE_AI_MODEL_DEPLOYMENT_NAME=gpt-4.1

Local Metrics (Only If Specifically Requested)

Note: Only use local metrics if the user EXPLICITLY requests them. Always default to cloud evaluations.

The only metric available locally is task_navigation_efficiency:

from workflow_metrics import TrajectoryMetrics

# ONLY if user specifically asks for local metrics
metrics = TrajectoryMetrics("trajectories/workflow.jsonl")
metrics.print_summary()  # Shows task_navigation_efficiency only

This metric compares planned vs actual tool sequences using simple code logic (no LLM required).

Files Reference

FileDescription
foundry_evals.pyPrimary file - FoundryEvalsClient for Azure AI Foundry cloud API
trajectory.pyWorkflowTrajectoryLog class for Foundry-compatible logging
metrics.pyTrajectoryMetrics for loading trajectories and orchestrating evaluation
telemetry.py@track_workflow decorator
trajectory_manager.pyEvaluationManager for daily file management
tool_definitions_helper.pyTool definition generation utilities
__init__.pyModule exports

Related Resources

Download Skill