AI-103 CERTIFICATION BOOTCAMP

# Inside the loop over response.output in your Session 13 function-calling agent.
# TOOLS maps tool names to your Python functions; these names are illustrative.
CONSEQUENTIAL = {"issue_refund", "close_account"}

if item.type == "function_call":
    args = json.loads(item.arguments)
    if item.name in CONSEQUENTIAL:
        print(f"{item.name} wants to run with {args}")
        if input("Approve? (y/N): ").strip().lower() == "y":
            result = TOOLS[item.name](**args)
        else:
            result = {"status": "rejected", "reason": "A reviewer declined this action."}
    else:
        result = TOOLS[item.name](**args)       # read-only tools run freely
    input_list.append(FunctionCallOutput(
        type="function_call_output",
        call_id=item.call_id,
        output=json.dumps(result),
    ))

# The packages Learn lists for client-side tracing to Azure Monitor
# (with the course floor for azure-ai-projects, set in Session 9)
pip install "azure-ai-projects>=2.4.0" azure-identity opentelemetry-sdk \
    azure-core-tracing-opentelemetry azure-monitor-opentelemetry

# Content recording is off by default. Leave it off unless the data is test data.
# export OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true

import json, os
os.environ["AZURE_EXPERIMENTAL_ENABLE_GENAI_TRACING"] = "true"  # before instrument()
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import PromptAgentDefinition, FunctionTool
from azure.ai.projects.telemetry import AIProjectInstrumentor, trace_function
from azure.monitor.opentelemetry import configure_azure_monitor
from openai.types.responses.response_input_param import FunctionCallOutput
from opentelemetry import trace

project = AIProjectClient(endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
                          credential=DefaultAzureCredential())
configure_azure_monitor(
    connection_string=project.telemetry.get_application_insights_connection_string())
AIProjectInstrumentor().instrument()
openai = project.get_openai_client()        # obtained after instrumentation

@trace_function
def get_leave_balance(staff_id: str) -> dict:  # your Session 13 function
    return {"staff_id": staff_id, "days_remaining": 12}

tool = FunctionTool(name="get_leave_balance", description="Leave days left for a staff ID.",
    parameters={"type": "object", "properties": {"staff_id": {"type": "string"}},
                "required": ["staff_id"], "additionalProperties": False}, strict=True)
agent = project.agents.create_version(agent_name="traced-hr-agent",
    definition=PromptAgentDefinition(model=os.environ["FOUNDRY_MODEL_NAME"],
        instructions="Answer leave questions. Use get_leave_balance when given a staff ID.",
        tools=[tool]))
ref = {"agent_reference": {"name": agent.name, "id": agent.id, "type": "agent_reference"}}

with trace.get_tracer(__name__).start_as_current_span("leave-balance-scenario"):
    conversation = openai.conversations.create()
    r = openai.responses.create(input="How many leave days does E1042 have left?",
                                conversation=conversation.id, extra_body=ref)
    outputs = [FunctionCallOutput(type="function_call_output", call_id=i.call_id,
               output=json.dumps(get_leave_balance(**json.loads(i.arguments))))
               for i in r.output if i.type == "function_call"]
    r = openai.responses.create(input=outputs, conversation=conversation.id, extra_body=ref)
    print(r.output_text, "| output tokens:", r.usage.output_tokens)
project.agents.delete_version(agent_name=agent.name, agent_version=agent.version)

E1042 has 12 days of annual leave remaining. | output tokens: <a number>

import os, random, time
import openai
from openai import OpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider

token_provider = get_bearer_token_provider(DefaultAzureCredential(),
                                           "https://ai.azure.com/.default")
client = OpenAI(base_url=os.environ["FOUNDRY_OPENAI_BASE_URL"],
                api_key=token_provider,
                max_retries=0)  # our loop retries, so switch off the SDK's own

def with_backoff(func, initial_delay=1.0, base=2.0, max_retries=6):
    def wrapper(*args, **kwargs):
        delay, attempt = initial_delay, 0
        while True:
            try:
                return func(*args, **kwargs)
            except openai.RateLimitError as e:          # HTTP 429 only
                attempt += 1
                if attempt > max_retries:
                    raise RuntimeError(f"Gave up after {max_retries} retries") from e
                delay *= base * (1 + random.random())   # exponential, with jitter
                print(f"  429 received; retry {attempt} in {delay:.1f}s")
                time.sleep(delay)
    return wrapper

@with_backoff
def ask(prompt):
    return client.responses.create(model=os.environ["FOUNDRY_MODEL_NAME"],
                                   input=prompt, max_output_tokens=50)

for i in range(20):  # a small burst against a deliberately small deployment
    print(i, ask(f"Give one English word for the number {i}.").output_text)

0 zero
1 one
2 two
  429 received; retry 1 in 2.7s
3 three
...
19 nineteen

import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import (
    EvaluationRule, ContinuousEvaluationRuleAction,
    EvaluationRuleFilter, EvaluationRuleEventType,
)

project = AIProjectClient(endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
                          credential=DefaultAzureCredential())
openai = project.get_openai_client()
AGENT_NAME = "<your-agent-name>"   # an agent that exists in this project

eval_object = openai.evals.create(
    name="Continuous Evaluation",
    data_source_config={"type": "azure_ai_source", "scenario": "responses"},
    testing_criteria=[{"type": "azure_ai_evaluator", "name": "violence_detection",
                       "evaluator_name": "builtin.violence"}],
)
rule = project.evaluation_rules.create_or_update(
    id="ai103-continuous-eval",
    evaluation_rule=EvaluationRule(
        display_name="AI-103 continuous evaluation",
        description="Evaluates sampled responses from my agent",
        action=ContinuousEvaluationRuleAction(eval_id=eval_object.id,
                                              max_hourly_runs=10),
        event_type=EvaluationRuleEventType.RESPONSE_COMPLETED,
        filter=EvaluationRuleFilter(agent_name=AGENT_NAME),
        enabled=True,   # run again with False to switch the rule off
    ),
)
print(f"Rule {rule.id} runs eval {eval_object.id}")

# After you have sent the agent some traffic:
runs = openai.evals.runs.list(eval_id=eval_object.id, order="desc", limit=10)
if runs.data and runs.data[0].report_url:
    print(f"Latest report: {runs.data[0].report_url}")

Rule ai103-continuous-eval runs eval <eval-id>
Latest report: <a link to the evaluation report>

# 1. Every resource group, then every resource in the course group
az group list -o table
az resource list -g rg-ai103 -o table

# 2. Model deployments on the Foundry resource: delete these first
az cognitiveservices account deployment list \
    -n <your-foundry-resource-name> -g rg-ai103 -o table
az cognitiveservices account deployment delete \
    --deployment-name "<deployment-name>" \
    -n <your-foundry-resource-name> -g rg-ai103

# 3. Resource IDs, then delete any resource you no longer need
az resource list -g rg-ai103 --query "[].id" -o tsv
az resource delete --ids "<resource-id>"

# 4. Foundry resources deleted in the last 48 hours (soft-deleted)
az cognitiveservices account list-deleted -o table

# Only if you are finishing or pausing the course: everything at once
# az group delete -n rg-ai103

<a table of resource groups>
<a table of resources in rg-ai103: name, type, location>
<a table of deployments, or nothing if none remain>
<a table of soft-deleted Foundry resources, or nothing>