mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
.NET: Python: Azure Functions feature branch (#1916)
* Python: Add Scaffolding for Durable AzureFunctions package to Agent Framework (#1823) * Add scafolding * update readme * add code owners and label * update owners * .NET: Durable extension: initial src and unit tests (#1900) * Python: Add Durable Agent Wrapper code (#1913) * add initial changes * Move code and add single sample * Update logger * Remove unused code * address PR comments * cleanup code and address comments --------- Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com> * Azure Functions .NET samples (#1939) * Python: Add Unit tests for Azurefunctions package (#1976) * Add Unit tests for Azurefunctions * remove duplicate import * .NET: [Feature Branch] Migrate state schema updates and support for agents as MCP tools (#1979) * Python: Add more samples for Azure Functions (#1980) * Move all samples * fix comments * remove dead lines * Make samples simpler * .NET: [Feature Branch] Durable Task extension integration tests (#2017) * .NET: [Feature Branch] Update OpenAI config for integration tests (#2063) * Python: Add Integration tests for AzureFunctions (#2020) * Add Integration tests * Remove DTS extension * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Add pyi file for type safety * Add samples in readme * Updated all readme instructions * Address comments * Update readmes * Fix requirements * Address comments --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * .NET: [Feature Branch] Update dotnet-build-and-test.yml to support integration tests (#2070) Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Fix DTS startup issue and improve logging (#2103) * .NET: [Feature Branch] Introduce Azure OpenAI config for .NET pipeline (#2106) Also fixes an issue where we were trying to start docker containers for integration tests on Windows, which doesn't work. Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Fix uv.lock after merge * Python: Add README for Azure Functions samples setup (#2100) * Add README for Azure Functions samples setup Added setup instructions for Azure Functions samples, including environment setup, virtual environment creation, and running samples. * Update python/samples/getting_started/azure_functions/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestion from @Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Laveesh Rohra <larohra@microsoft.com> * Fix or remove broken markdown file links (#2115) * .NET: [Feature Branch] Update HTTP API to be consistent across languages (#2118) * Python: Fix AzureFunctions Integration Tests (#2116) * Add Identity Auth to samples * Update python/samples/getting_started/azure_functions/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/samples/getting_started/azure_functions/01_single_agent/function_app.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/samples/getting_started/azure_functions/02_multi_agent/function_app.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/samples/getting_started/azure_functions/06_multi_agent_orchestration_conditionals/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Python: Fix Http Schema (#2112) * Rename to threadid * Respond in plain text * Make snake-case * Add http prefix * rename to wait-for-response * Add query param check * address comments * .NET: Remove IsPackable=false in preparation for nuget release (#2142) * Python: Move `azurefunctions` to `azure` for import (#2141) * Move import to Azure * fix mypy * Update python/packages/azurefunctions/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Add missing types * Address comments --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/packages/azurefunctions/pyproject.toml Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/packages/azurefunctions/agent_framework_azurefunctions/__init__.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Fix imports * Address PR feedback from westey-m (#2150) - Adds a link from the /dotnet/samples/README.md to /dotnet/samples/AzureFunctions - Make DurableAgentThread deserialization internal for future-proofing - Update JSON serialization logic to address recently discovered issues with source generator serialization * Address comments (#2160) --------- Co-authored-by: Laveesh Rohra <larohra@microsoft.com> Co-authored-by: Chris Gillum <cgillum@microsoft.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Anirudh Garg <anirudhg@microsoft.com>
This commit is contained in:
Dmytro Struk
committed by
GitHub
Unverified
GitHub
parent
5537b1da79
commit
67a8147151
@@ -193,6 +193,18 @@ This directory contains samples demonstrating the capabilities of Microsoft Agen
|
||||
| [`getting_started/multimodal_input/azure_responses_multimodal.py`](./getting_started/multimodal_input/azure_responses_multimodal.py) | Azure OpenAI Responses with multimodal (image) input example |
|
||||
| [`getting_started/multimodal_input/openai_chat_multimodal.py`](./getting_started/multimodal_input/openai_chat_multimodal.py) | OpenAI Chat with multimodal (image) input example |
|
||||
|
||||
## Azure Functions
|
||||
|
||||
| Sample | Description |
|
||||
|--------|-------------|
|
||||
| [`getting_started/azure_functions/01_single_agent/`](./getting_started/azure_functions/01_single_agent/) | Host a single agent in Azure Functions with Durable Extension HTTP endpoints and per-session state. |
|
||||
| [`getting_started/azure_functions/02_multi_agent/`](./getting_started/azure_functions/02_multi_agent/) | Register multiple agents in one function app with dedicated run routes and a health check endpoint. |
|
||||
| [`getting_started/azure_functions/03_callbacks/`](./getting_started/azure_functions/03_callbacks/) | Capture streaming response telemetry via Durable Extension callbacks exposed through HTTP APIs. |
|
||||
| [`getting_started/azure_functions/04_single_agent_orchestration_chaining/`](./getting_started/azure_functions/04_single_agent_orchestration_chaining/) | Chain sequential agent executions inside a durable orchestration while preserving the shared thread context. |
|
||||
| [`getting_started/azure_functions/05_multi_agent_orchestration_concurrency/`](./getting_started/azure_functions/05_multi_agent_orchestration_concurrency/) | Run two agents concurrently within a durable orchestration and combine their domain-specific outputs. |
|
||||
| [`getting_started/azure_functions/06_multi_agent_orchestration_conditionals/`](./getting_started/azure_functions/06_multi_agent_orchestration_conditionals/) | Route orchestration logic based on structured agent responses for spam detection and reply drafting. |
|
||||
| [`getting_started/azure_functions/07_single_agent_orchestration_hitl/`](./getting_started/azure_functions/07_single_agent_orchestration_hitl/) | Implement a human-in-the-loop approval loop that iterates on agent output inside a durable orchestration. |
|
||||
|
||||
## Observability
|
||||
|
||||
| File | Description |
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
# Single Agent Sample (Python)
|
||||
|
||||
This sample demonstrates how to use the Durable Extension for Agent Framework to create a simple Azure Functions app that hosts a single AI agent and provides direct HTTP API access for interactive conversations.
|
||||
|
||||
## Key Concepts Demonstrated
|
||||
|
||||
- Defining a simple agent with the Microsoft Agent Framework and wiring it into
|
||||
an Azure Functions app via the Durable Extension for Agent Framework.
|
||||
- Calling the agent through generated HTTP endpoints (`/api/agents/Joker/run`).
|
||||
- Managing conversation state with thread identifiers, so multiple clients can
|
||||
interact with the agent concurrently without sharing context.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Follow the common setup steps in `../README.md` to install tooling, configure Azure OpenAI credentials, and install the Python dependencies for this sample.
|
||||
|
||||
## Running the Sample
|
||||
|
||||
Send a prompt to the Joker agent:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/agents/Joker/run \
|
||||
-H "Content-Type: text/plain" \
|
||||
-d "Tell me a short joke about cloud computing."
|
||||
```
|
||||
|
||||
The agent responds with a JSON payload that includes the generated joke.
|
||||
|
||||
> **Note:** To return immediately with an HTTP 202 response instead of waiting for the agent output, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body. The default behavior waits for the response.
|
||||
|
||||
## Expected Output
|
||||
|
||||
When you send a POST request with plain-text input, the Functions host responds with an HTTP 202 and queues the request for the durable agent entity. A typical response body looks like the following:
|
||||
Expected HTTP 202 payload:
|
||||
|
||||
```json
|
||||
{
|
||||
"status": "accepted",
|
||||
"response": "Agent request accepted",
|
||||
"message": "Tell me a short joke about cloud computing.",
|
||||
"thread_id": "<guid>",
|
||||
"correlation_id": "<guid>"
|
||||
}
|
||||
```
|
||||
@@ -0,0 +1,22 @@
|
||||
### Joker Agent Sample Interactions
|
||||
@baseUrl = http://localhost:7071
|
||||
@agentName = Joker
|
||||
@agentRoute = {{baseUrl}}/api/agents/{{agentName}}
|
||||
@healthRoute = {{baseUrl}}/api/health
|
||||
|
||||
### Health Check
|
||||
GET {{healthRoute}}
|
||||
|
||||
### Ask for a joke (JSON payload)
|
||||
POST {{agentRoute}}/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"message": "Add a security element to it.",
|
||||
"thread_id": "thread-001"
|
||||
}
|
||||
|
||||
### Ask for a joke (plain text payload)
|
||||
POST {{agentRoute}}/run
|
||||
|
||||
Give me a programming joke about race conditions.
|
||||
@@ -0,0 +1,37 @@
|
||||
"""Host a single Azure OpenAI-powered agent inside Azure Functions.
|
||||
|
||||
Components used in this sample:
|
||||
- AzureOpenAIChatClient to call the Azure OpenAI chat deployment.
|
||||
- AgentFunctionApp to expose HTTP endpoints via the Durable Functions extension.
|
||||
|
||||
Prerequisites: set `AZURE_OPENAI_ENDPOINT` and `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME` (plus `AZURE_OPENAI_API_KEY` or Azure CLI authentication) before starting the Functions host."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from agent_framework.azure import AgentFunctionApp, AzureOpenAIChatClient
|
||||
from azure.identity import AzureCliCredential
|
||||
# 1. Instantiate the agent with the chosen deployment and instructions.
|
||||
def _create_agent() -> Any:
|
||||
"""Create the Joker agent."""
|
||||
|
||||
return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
|
||||
name="Joker",
|
||||
instructions="You are good at telling jokes.",
|
||||
)
|
||||
|
||||
|
||||
# 2. Register the agent with AgentFunctionApp so Azure Functions exposes the required triggers.
|
||||
app = AgentFunctionApp(agents=[_create_agent()], enable_health_check=True, max_poll_retries=50)
|
||||
|
||||
"""
|
||||
Expected output when invoking `POST /api/agents/Joker/run` with plain-text input:
|
||||
|
||||
HTTP/1.1 202 Accepted
|
||||
{
|
||||
"status": "accepted",
|
||||
"response": "Agent request accepted",
|
||||
"message": "Tell me a short joke about cloud computing.",
|
||||
"conversation_id": "<guid>",
|
||||
"correlation_id": "<guid>"
|
||||
}
|
||||
"""
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"extensionBundle": {
|
||||
"id": "Microsoft.Azure.Functions.ExtensionBundle",
|
||||
"version": "[4.*, 5.0.0)"
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "%TASKHUB_NAME%"
|
||||
}
|
||||
}
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"IsEncrypted": false,
|
||||
"Values": {
|
||||
"FUNCTIONS_WORKER_RUNTIME": "python",
|
||||
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
|
||||
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
|
||||
"TASKHUB_NAME": "default",
|
||||
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
|
||||
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "<AZURE_OPENAI_CHAT_DEPLOYMENT_NAME>",
|
||||
"AZURE_OPENAI_API_KEY": "<AZURE_OPENAI_API_KEY>"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
agent-framework-azurefunctions
|
||||
azure-identity
|
||||
@@ -0,0 +1,104 @@
|
||||
# Multi-Agent Sample
|
||||
|
||||
This sample demonstrates how to use the Durable Extension for Agent Framework to create an Azure Functions app that hosts multiple AI agents and provides direct HTTP API access for interactive conversations with each agent.
|
||||
|
||||
## Key Concepts Demonstrated
|
||||
|
||||
- Using the Microsoft Agent Framework to define multiple AI agents with unique names and instructions.
|
||||
- Registering multiple agents with the Function app and running them using HTTP.
|
||||
- Conversation management (via thread IDs) for isolated interactions per agent.
|
||||
- Two different methods for registering agents: list-based initialization and incremental addition.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Complete the common environment preparation steps described in `../README.md`, including installing Azure Functions Core Tools, starting Azurite, configuring Azure OpenAI settings, and installing this sample's requirements.
|
||||
|
||||
## Running the Sample
|
||||
|
||||
With the environment setup and function app running, you can test the sample by sending HTTP requests to the different agent endpoints.
|
||||
|
||||
You can use the `demo.http` file to send messages to the agents, or a command line tool like `curl` as shown below:
|
||||
|
||||
> **Note:** Each endpoint waits for the agent response by default. To receive an immediate HTTP 202 instead, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body.
|
||||
|
||||
### Test the Weather Agent
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
Weather agent request:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/agents/WeatherAgent/run \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"message": "What is the weather in Seattle?"}'
|
||||
```
|
||||
|
||||
Expected HTTP 202 payload:
|
||||
|
||||
```json
|
||||
{
|
||||
"status": "accepted",
|
||||
"response": "Agent request accepted",
|
||||
"message": "What is the weather in Seattle?",
|
||||
"thread_id": "<guid>",
|
||||
"correlation_id": "<guid>"
|
||||
}
|
||||
```
|
||||
|
||||
Math agent request:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/agents/MathAgent/run \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"message": "Calculate a 20% tip on a $50 bill"}'
|
||||
```
|
||||
|
||||
Expected HTTP 202 payload:
|
||||
|
||||
```json
|
||||
{
|
||||
"status": "accepted",
|
||||
"response": "Agent request accepted",
|
||||
"message": "Calculate a 20% tip on a $50 bill",
|
||||
"thread_id": "<guid>",
|
||||
"correlation_id": "<guid>"
|
||||
}
|
||||
```
|
||||
|
||||
Health check (optional):
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/health
|
||||
```
|
||||
|
||||
Expected response:
|
||||
|
||||
```json
|
||||
{
|
||||
"status": "healthy",
|
||||
"agents": [
|
||||
{"name": "WeatherAgent", "type": "ChatAgent"},
|
||||
{"name": "MathAgent", "type": "ChatAgent"}
|
||||
],
|
||||
"agent_count": 2
|
||||
}
|
||||
```
|
||||
|
||||
## Code Structure
|
||||
|
||||
The sample demonstrates two ways to register multiple agents:
|
||||
|
||||
### Option 1: Pass list of agents during initialization
|
||||
```python
|
||||
app = AgentFunctionApp(agents=[weather_agent, math_agent])
|
||||
```
|
||||
|
||||
### Option 2: Add agents incrementally (commented in sample)
|
||||
```python
|
||||
app = AgentFunctionApp()
|
||||
app.add_agent(weather_agent)
|
||||
app.add_agent(math_agent)
|
||||
```
|
||||
|
||||
Each agent automatically gets:
|
||||
- `POST /api/agents/{agent_name}/run` - Send messages to the agent
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
### DAFx Multi-Agent Function App - HTTP Samples
|
||||
### Use with the VS Code REST Client extension or any HTTP client
|
||||
###
|
||||
### API Structure:
|
||||
### - POST /api/agents/{agentName}/run -> Send a message to an agent
|
||||
### - GET /api/health -> Health check and agent metadata
|
||||
|
||||
### Variables
|
||||
@baseUrl = http://localhost:7071
|
||||
@weatherAgentName = WeatherAgent
|
||||
@mathAgentName = MathAgent
|
||||
@weatherAgentRoute = {{baseUrl}}/api/agents/{{weatherAgentName}}
|
||||
@mathAgentRoute = {{baseUrl}}/api/agents/{{mathAgentName}}
|
||||
@healthRoute = {{baseUrl}}/api/health
|
||||
|
||||
### Health Check
|
||||
# Confirms the Azure Functions app is running and both agents are registered
|
||||
# Expected response:
|
||||
# {
|
||||
# "status": "healthy",
|
||||
# "agents": [
|
||||
# {"name": "WeatherAgent", "type": "AzureOpenAIAssistantsAgent"},
|
||||
# {"name": "MathAgent", "type": "AzureOpenAIAssistantsAgent"}
|
||||
# ],
|
||||
# "agent_count": 2
|
||||
# }
|
||||
GET {{healthRoute}}
|
||||
|
||||
###
|
||||
|
||||
### Weather Agent - Current Conditions
|
||||
# Tests the Weather agent's tool-assisted response path
|
||||
# Expected response: { "response": "The weather in Seattle...", "status": "success" }
|
||||
POST {{weatherAgentRoute}}/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"message": "What is the weather in Seattle?",
|
||||
"thread_id": "weather-user-001"
|
||||
}
|
||||
|
||||
###
|
||||
|
||||
|
||||
### Math Agent - Tip Calculation
|
||||
# Exercises the Math agent with a calculation request
|
||||
# Expected response: { "response": "A 20% tip on a $50 bill is $10...", "status": "success" }
|
||||
POST {{mathAgentRoute}}/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"message": "Calculate a 20% tip on a $50 bill",
|
||||
"thread_id": "math-user-001"
|
||||
}
|
||||
|
||||
###
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""Host multiple Azure OpenAI agents inside a single Azure Functions app.
|
||||
|
||||
Components used in this sample:
|
||||
- AzureOpenAIChatClient to create agents bound to a shared Azure OpenAI deployment.
|
||||
- AgentFunctionApp to register multiple agents and expose dedicated HTTP endpoints.
|
||||
- Custom tool functions to demonstrate tool invocation from different agents.
|
||||
|
||||
Prerequisites: set `AZURE_OPENAI_ENDPOINT` and `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME`, plus either
|
||||
`AZURE_OPENAI_API_KEY` or authenticate with Azure CLI before starting the Functions host."""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from agent_framework.azure import AgentFunctionApp, AzureOpenAIChatClient
|
||||
from azure.identity import AzureCliCredential
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_weather(location: str) -> dict[str, Any]:
|
||||
"""Get current weather for a location."""
|
||||
|
||||
logger.info(f"🔧 [TOOL CALLED] get_weather(location={location})")
|
||||
result = {
|
||||
"location": location,
|
||||
"temperature": 72,
|
||||
"conditions": "Sunny",
|
||||
"humidity": 45,
|
||||
}
|
||||
logger.info(f"✓ [TOOL RESULT] {result}")
|
||||
return result
|
||||
|
||||
|
||||
def calculate_tip(bill_amount: float, tip_percentage: float = 15.0) -> dict[str, Any]:
|
||||
"""Calculate tip amount and total bill."""
|
||||
|
||||
logger.info(
|
||||
f"🔧 [TOOL CALLED] calculate_tip(bill_amount={bill_amount}, tip_percentage={tip_percentage})"
|
||||
)
|
||||
tip = bill_amount * (tip_percentage / 100)
|
||||
total = bill_amount + tip
|
||||
result = {
|
||||
"bill_amount": bill_amount,
|
||||
"tip_percentage": tip_percentage,
|
||||
"tip_amount": round(tip, 2),
|
||||
"total": round(total, 2),
|
||||
}
|
||||
logger.info(f"✓ [TOOL RESULT] {result}")
|
||||
return result
|
||||
|
||||
|
||||
# 1. Create multiple agents, each with its own instruction set and tools.
|
||||
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
|
||||
|
||||
weather_agent = chat_client.create_agent(
|
||||
name="WeatherAgent",
|
||||
instructions="You are a helpful weather assistant. Provide current weather information.",
|
||||
tools=[get_weather],
|
||||
)
|
||||
|
||||
math_agent = chat_client.create_agent(
|
||||
name="MathAgent",
|
||||
instructions="You are a helpful math assistant. Help users with calculations like tip calculations.",
|
||||
tools=[calculate_tip],
|
||||
)
|
||||
|
||||
|
||||
# 2. Register both agents with AgentFunctionApp to expose their HTTP routes and health check.
|
||||
app = AgentFunctionApp(agents=[weather_agent, math_agent], enable_health_check=True, max_poll_retries=50)
|
||||
|
||||
# Option 2: Add agents after initialization (commented out as we're using Option 1)
|
||||
# app = AgentFunctionApp(enable_health_check=True)
|
||||
# app.add_agent(weather_agent)
|
||||
# app.add_agent(math_agent)
|
||||
|
||||
"""
|
||||
Expected output when invoking `POST /api/agents/WeatherAgent/run`:
|
||||
|
||||
HTTP/1.1 202 Accepted
|
||||
{
|
||||
"status": "accepted",
|
||||
"response": "Agent request accepted",
|
||||
"message": "What is the weather in Seattle?",
|
||||
"conversation_id": "<guid>",
|
||||
"correlation_id": "<guid>"
|
||||
}
|
||||
|
||||
Expected output when invoking `POST /api/agents/MathAgent/run`:
|
||||
|
||||
HTTP/1.1 202 Accepted
|
||||
{
|
||||
"status": "accepted",
|
||||
"response": "Agent request accepted",
|
||||
"message": "Calculate a 20% tip on a $50 bill",
|
||||
"conversation_id": "<guid>",
|
||||
"correlation_id": "<guid>"
|
||||
}
|
||||
"""
|
||||
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"logging": {
|
||||
"applicationInsights": {
|
||||
"samplingSettings": {
|
||||
"isEnabled": true,
|
||||
"maxTelemetryItemsPerSecond": 20
|
||||
}
|
||||
}
|
||||
},
|
||||
"extensionBundle": {
|
||||
"id": "Microsoft.Azure.Functions.ExtensionBundle",
|
||||
"version": "[4.*, 5.0.0)"
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "%TASKHUB_NAME%"
|
||||
}
|
||||
}
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"IsEncrypted": false,
|
||||
"Values": {
|
||||
"FUNCTIONS_WORKER_RUNTIME": "python",
|
||||
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
|
||||
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
|
||||
"TASKHUB_NAME": "default",
|
||||
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
|
||||
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "<AZURE_OPENAI_CHAT_DEPLOYMENT_NAME>",
|
||||
"AZURE_OPENAI_API_KEY": "<AZURE_OPENAI_API_KEY>"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
agent-framework-azurefunctions
|
||||
azure-identity
|
||||
@@ -0,0 +1,82 @@
|
||||
# Callback Telemetry Sample
|
||||
|
||||
This sample demonstrates how to use the Durable Extension for Agent Framework's response callbacks to observe
|
||||
streaming updates and final agent responses in real time. The `ConversationAuditTrail` callback
|
||||
records each chunk received from the Azure OpenAI agent and exposes the collected events through
|
||||
an HTTP API that can be polled by a web client or dashboard.
|
||||
|
||||
## Highlights
|
||||
|
||||
- Registers a default `AgentResponseCallbackProtocol` implementation that logs streaming and final
|
||||
responses.
|
||||
- Persists callback events in an in-memory store and exposes them via
|
||||
`GET /api/agents/{agentName}/callbacks/{thread_id}`.
|
||||
- Shows how to reset stored callback events with `DELETE /api/agents/{agentName}/callbacks/{thread_id}`.
|
||||
- Works alongside the standard `/api/agents/{agentName}/run` endpoint so you can correlate callback
|
||||
telemetry with agent responses.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Complete the shared environment setup steps in `../README.md`, including creating a virtual environment, installing dependencies, and configuring Azure OpenAI credentials and storage settings.
|
||||
|
||||
> **Note:** The sample stores callback events in memory for simplicity. For production scenarios you
|
||||
> should persist events to Application Insights, Azure Storage, Cosmos DB, or another durable store.
|
||||
|
||||
## Running the Sample
|
||||
|
||||
Send a prompt to the agent:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/agents/CallbackAgent/run \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"message": "Tell me a short joke"}'
|
||||
```
|
||||
|
||||
> **Note:** The run endpoint waits for the agent response by default. To return immediately, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body.
|
||||
|
||||
Poll callback telemetry (replace `<conversationId>` with the value from the POST response):
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/agents/CallbackAgent/callbacks/<conversationId>
|
||||
```
|
||||
|
||||
Reset stored events:
|
||||
|
||||
```bash
|
||||
curl -X DELETE http://localhost:7071/api/agents/CallbackAgent/callbacks/<conversationId>
|
||||
```
|
||||
|
||||
## Expected Output
|
||||
|
||||
When you call `GET /api/agents/CallbackAgent/callbacks/{thread_id}` after sending a request to the agent,
|
||||
the API returns a list of streaming and final callback events similar to the following:
|
||||
|
||||
```json
|
||||
[
|
||||
{
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"agent_name": "CallbackAgent",
|
||||
"thread_id": "<thread_id>",
|
||||
"correlation_id": "<guid>",
|
||||
"request_message": "Tell me a short joke",
|
||||
"event_type": "stream",
|
||||
"update_kind": "text",
|
||||
"text": "Sure, here's a joke..."
|
||||
},
|
||||
{
|
||||
"timestamp": "2024-01-01T00:00:01Z",
|
||||
"agent_name": "CallbackAgent",
|
||||
"thread_id": "<thread_id>",
|
||||
"correlation_id": "<guid>",
|
||||
"request_message": "Tell me a short joke",
|
||||
"event_type": "final",
|
||||
"response_text": "Why did the cloud...",
|
||||
"usage": {
|
||||
"type": "usage_details",
|
||||
"input_token_count": 159,
|
||||
"output_token_count": 29,
|
||||
"total_token_count": 188
|
||||
}
|
||||
}
|
||||
]
|
||||
```
|
||||
@@ -0,0 +1,30 @@
|
||||
### Callback Sample - API Tests
|
||||
### Use with VS Code REST Client or another HTTP testing tool.
|
||||
###
|
||||
### Endpoints introduced in this sample:
|
||||
### - POST /api/agents/{agentName}/run : send a message to the agent
|
||||
### - GET /api/agents/{agentName}/callbacks/{thread_id} : retrieve callback telemetry
|
||||
### - DELETE /api/agents/{agentName}/callbacks/{thread_id} : clear stored callback events
|
||||
|
||||
@baseUrl = http://localhost:7071
|
||||
@agentName = CallbackAgent
|
||||
@agentRoute = {{baseUrl}}/api/agents/{{agentName}}
|
||||
@thread_id = test-thread-00
|
||||
|
||||
### Health Check
|
||||
GET {{baseUrl}}/api/health
|
||||
|
||||
### Send message (callbacks will capture streaming + final response)
|
||||
POST {{agentRoute}}/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"message": "Generate a short weather update for Paris and mention streaming callbacks.",
|
||||
"thread_id": "{{thread_id}}"
|
||||
}
|
||||
|
||||
### Inspect callback telemetry
|
||||
GET {{agentRoute}}/callbacks/{{thread_id}}
|
||||
|
||||
### Clear stored callback telemetry for the thread
|
||||
DELETE {{agentRoute}}/callbacks/{{thread_id}}
|
||||
@@ -0,0 +1,183 @@
|
||||
"""Capture agent response callbacks inside Azure Functions.
|
||||
|
||||
Components used in this sample:
|
||||
- AzureOpenAIChatClient to build an agent that streams interim updates.
|
||||
- AgentFunctionApp with a default AgentResponseCallbackProtocol implementation.
|
||||
- Azure Functions HTTP triggers that expose callback telemetry via REST.
|
||||
|
||||
Prerequisites: set `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME`, and either
|
||||
`AZURE_OPENAI_API_KEY` or authenticate with Azure CLI before starting the Functions host."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, DefaultDict
|
||||
|
||||
import azure.functions as func
|
||||
from agent_framework import AgentRunResponseUpdate
|
||||
from agent_framework.azure import (
|
||||
AgentCallbackContext,
|
||||
AgentFunctionApp,
|
||||
AgentResponseCallbackProtocol,
|
||||
AzureOpenAIChatClient,
|
||||
)
|
||||
from azure.identity import AzureCliCredential
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# 1. Maintain an in-memory store for callback events keyed by thread ID (replace with durable storage in production).
|
||||
CallbackStore = DefaultDict[str, list[dict[str, Any]]]
|
||||
callback_events: CallbackStore = defaultdict(list)
|
||||
|
||||
|
||||
def _serialize_usage(usage: Any) -> Any:
|
||||
"""Best-effort serialization for agent usage metadata."""
|
||||
|
||||
if usage is None:
|
||||
return None
|
||||
|
||||
model_dump = getattr(usage, "model_dump", None)
|
||||
if callable(model_dump):
|
||||
return model_dump()
|
||||
|
||||
to_dict = getattr(usage, "to_dict", None)
|
||||
if callable(to_dict):
|
||||
return to_dict()
|
||||
|
||||
return str(usage)
|
||||
|
||||
|
||||
class ConversationAuditTrail(AgentResponseCallbackProtocol):
|
||||
"""Callback that records streaming chunks and final responses for later inspection."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._logger = logging.getLogger("durableagent.samples.callbacks.audit")
|
||||
|
||||
async def on_streaming_response_update(
|
||||
self,
|
||||
update: AgentRunResponseUpdate,
|
||||
context: AgentCallbackContext,
|
||||
) -> None:
|
||||
event = self._build_base_event(context)
|
||||
event.update(
|
||||
{
|
||||
"event_type": "stream",
|
||||
"update_kind": getattr(update, "kind", "text"),
|
||||
"text": getattr(update, "text", None),
|
||||
}
|
||||
)
|
||||
thread_id = context.thread_id or ""
|
||||
callback_events[thread_id].append(event)
|
||||
|
||||
preview = event.get("text") or event.get("update_kind")
|
||||
self._logger.info(
|
||||
"[%s][%s] streaming chunk: %s",
|
||||
context.agent_name,
|
||||
context.correlation_id,
|
||||
preview,
|
||||
)
|
||||
|
||||
async def on_agent_response(self, response, context: AgentCallbackContext) -> None:
|
||||
event = self._build_base_event(context)
|
||||
event.update(
|
||||
{
|
||||
"event_type": "final",
|
||||
"response_text": getattr(response, "text", None),
|
||||
"usage": _serialize_usage(getattr(response, "usage_details", None)),
|
||||
}
|
||||
)
|
||||
thread_id = context.thread_id or ""
|
||||
callback_events[thread_id].append(event)
|
||||
|
||||
self._logger.info(
|
||||
"[%s][%s] final response recorded",
|
||||
context.agent_name,
|
||||
context.correlation_id,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _build_base_event(context: AgentCallbackContext) -> dict[str, Any]:
|
||||
thread_id = context.thread_id
|
||||
return {
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"agent_name": context.agent_name,
|
||||
"thread_id": thread_id,
|
||||
"correlation_id": context.correlation_id,
|
||||
"request_message": context.request_message,
|
||||
}
|
||||
|
||||
|
||||
# 2. Create the agent that will emit streaming updates and final responses.
|
||||
callback_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
|
||||
name="CallbackAgent",
|
||||
instructions=(
|
||||
"You are a friendly assistant that narrates actions while responding. "
|
||||
"Keep answers concise and acknowledge when callbacks capture streaming updates."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# 3. Register the agent inside AgentFunctionApp with a default callback instance.
|
||||
audit_callback = ConversationAuditTrail()
|
||||
app = AgentFunctionApp(enable_health_check=True, default_callback=audit_callback)
|
||||
app.add_agent(callback_agent)
|
||||
|
||||
|
||||
@app.function_name("get_callback_events")
|
||||
@app.route(route="agents/{agent_name}/callbacks/{thread_id}", methods=["GET"])
|
||||
async def get_callback_events(req: func.HttpRequest) -> func.HttpResponse:
|
||||
"""Return all callback events collected for a thread."""
|
||||
|
||||
thread_id = req.route_params.get("thread_id", "")
|
||||
events = callback_events.get(thread_id, [])
|
||||
return func.HttpResponse(
|
||||
json.dumps(events, indent=2),
|
||||
status_code=200,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
@app.function_name("reset_callback_events")
|
||||
@app.route(route="agents/{agent_name}/callbacks/{thread_id}", methods=["DELETE"])
|
||||
async def reset_callback_events(req: func.HttpRequest) -> func.HttpResponse:
|
||||
"""Clear the stored callback events for a thread."""
|
||||
|
||||
thread_id = req.route_params.get("thread_id", "")
|
||||
callback_events.pop(thread_id, None)
|
||||
return func.HttpResponse(status_code=204)
|
||||
|
||||
|
||||
"""
|
||||
Expected output when querying `GET /api/agents/CallbackAgent/callbacks/{thread_id}`:
|
||||
|
||||
HTTP/1.1 200 OK
|
||||
[
|
||||
{
|
||||
"timestamp": "2024-01-01T00:00:00Z",
|
||||
"agent_name": "CallbackAgent",
|
||||
"thread_id": "<thread_id>",
|
||||
"correlation_id": "<guid>",
|
||||
"request_message": "Tell me a short joke",
|
||||
"event_type": "stream",
|
||||
"update_kind": "text",
|
||||
"text": "Sure, here's a joke..."
|
||||
},
|
||||
{
|
||||
"timestamp": "2024-01-01T00:00:01Z",
|
||||
"agent_name": "CallbackAgent",
|
||||
"thread_id": "<thread_id>",
|
||||
"correlation_id": "<guid>",
|
||||
"request_message": "Tell me a short joke",
|
||||
"event_type": "final",
|
||||
"response_text": "Why did the cloud...",
|
||||
"usage": {
|
||||
"type": "usage_details",
|
||||
"input_token_count": 159,
|
||||
"output_token_count": 29,
|
||||
"total_token_count": 188
|
||||
}
|
||||
}
|
||||
]
|
||||
"""
|
||||
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"logging": {
|
||||
"applicationInsights": {
|
||||
"samplingSettings": {
|
||||
"isEnabled": true,
|
||||
"maxTelemetryItemsPerSecond": 20
|
||||
}
|
||||
}
|
||||
},
|
||||
"extensionBundle": {
|
||||
"id": "Microsoft.Azure.Functions.ExtensionBundle",
|
||||
"version": "[4.*, 5.0.0)"
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "%TASKHUB_NAME%"
|
||||
}
|
||||
}
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"IsEncrypted": false,
|
||||
"Values": {
|
||||
"FUNCTIONS_WORKER_RUNTIME": "python",
|
||||
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
|
||||
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
|
||||
"TASKHUB_NAME": "default",
|
||||
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
|
||||
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "<AZURE_OPENAI_CHAT_DEPLOYMENT_NAME>",
|
||||
"AZURE_OPENAI_API_KEY": "<AZURE_OPENAI_API_KEY>"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
agent-framework-azurefunctions
|
||||
azure-identity
|
||||
+53
@@ -0,0 +1,53 @@
|
||||
# Single Agent Orchestration Sample (Python)
|
||||
|
||||
This sample shows how to chain two invocations of the same agent inside a Durable Functions orchestration while
|
||||
preserving the conversation state between runs.
|
||||
|
||||
## Key Concepts
|
||||
- Deterministic orchestrations that make sequential agent calls on a shared thread
|
||||
- Reusing an agent thread to carry conversation history across invocations
|
||||
- HTTP endpoints for starting the orchestration and polling for status/output
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Start with the shared setup instructions in `../README.md` to create a virtual environment, install dependencies, and configure Azure OpenAI and storage settings.
|
||||
|
||||
## Running the Sample
|
||||
Start the orchestration:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/singleagent/run
|
||||
```
|
||||
|
||||
Poll the returned `statusQueryGetUri` until completion:
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/singleagent/status/<instanceId>
|
||||
```
|
||||
|
||||
> **Note:** The underlying agent run endpoint now waits for responses by default. If you invoke it directly and prefer an immediate HTTP 202, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the payload.
|
||||
|
||||
The orchestration first requests an inspirational sentence from the agent, then refines the initial response while
|
||||
keeping it under 25 words—mirroring the behaviour of the corresponding .NET sample.
|
||||
|
||||
## Expected Output
|
||||
|
||||
Sample response when starting the orchestration:
|
||||
|
||||
```json
|
||||
{
|
||||
"message": "Single-agent orchestration started.",
|
||||
"instanceId": "ebb5c1df123e4d6fb8e7d703ffd0d0b0",
|
||||
"statusQueryGetUri": "http://localhost:7071/api/singleagent/status/ebb5c1df123e4d6fb8e7d703ffd0d0b0"
|
||||
}
|
||||
```
|
||||
|
||||
Sample completed status payload:
|
||||
|
||||
```json
|
||||
{
|
||||
"instanceId": "ebb5c1df123e4d6fb8e7d703ffd0d0b0",
|
||||
"runtimeStatus": "Completed",
|
||||
"output": "Learning is a journey where curiosity turns effort into mastery."
|
||||
}
|
||||
```
|
||||
+9
@@ -0,0 +1,9 @@
|
||||
### Start the single-agent orchestration
|
||||
POST http://localhost:7071/api/singleagent/run
|
||||
|
||||
|
||||
### Check the status of the orchestration
|
||||
|
||||
@instanceId =<Replace with the instance ID from the response above>
|
||||
|
||||
GET http://localhost:7071/api/singleagent/status/{{instanceId}}
|
||||
+174
@@ -0,0 +1,174 @@
|
||||
"""Chain two runs of a single agent inside a Durable Functions orchestration.
|
||||
|
||||
Components used in this sample:
|
||||
- AzureOpenAIChatClient to construct the writer agent hosted by Agent Framework.
|
||||
- AgentFunctionApp to surface HTTP and orchestration triggers via the Azure Functions extension.
|
||||
- Durable Functions orchestration to run sequential agent invocations on the same conversation thread.
|
||||
|
||||
Prerequisites: configure `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME`, and either
|
||||
`AZURE_OPENAI_API_KEY` or authenticate with Azure CLI before starting the Functions host."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import azure.durable_functions as df
|
||||
import azure.functions as func
|
||||
from agent_framework.azure import AgentFunctionApp, AzureOpenAIChatClient
|
||||
from azure.durable_functions import DurableOrchestrationContext
|
||||
from azure.identity import AzureCliCredential
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 1. Define the agent name used across the orchestration.
|
||||
WRITER_AGENT_NAME = "WriterAgent"
|
||||
|
||||
|
||||
# 2. Create the writer agent that will be invoked twice within the orchestration.
|
||||
def _create_writer_agent() -> Any:
|
||||
"""Create the writer agent with the same persona as the C# sample."""
|
||||
|
||||
instructions = (
|
||||
"You refine short pieces of text. When given an initial sentence you enhance it;\n"
|
||||
"when given an improved sentence you polish it further."
|
||||
)
|
||||
|
||||
return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
|
||||
name=WRITER_AGENT_NAME,
|
||||
instructions=instructions,
|
||||
)
|
||||
|
||||
|
||||
# 3. Register the agent with AgentFunctionApp so HTTP and orchestration triggers are exposed.
|
||||
app = AgentFunctionApp(agents=[_create_writer_agent()], enable_health_check=True)
|
||||
|
||||
|
||||
# 4. Orchestration that runs the agent sequentially on a shared thread for chaining behaviour.
|
||||
@app.orchestration_trigger(context_name="context")
|
||||
def single_agent_orchestration(context: DurableOrchestrationContext):
|
||||
"""Run the writer agent twice on the same thread to mirror chaining behaviour."""
|
||||
|
||||
writer = app.get_agent(context, WRITER_AGENT_NAME)
|
||||
writer_thread = writer.get_new_thread()
|
||||
|
||||
initial = yield writer.run(
|
||||
messages="Write a concise inspirational sentence about learning.",
|
||||
thread=writer_thread,
|
||||
)
|
||||
|
||||
improved_prompt = (
|
||||
"Improve this further while keeping it under 25 words: "
|
||||
f"{initial.get('response', '').strip()}"
|
||||
)
|
||||
|
||||
refined = yield writer.run(
|
||||
messages=improved_prompt,
|
||||
thread=writer_thread,
|
||||
)
|
||||
|
||||
return refined.get("response", "")
|
||||
|
||||
|
||||
# 5. HTTP endpoint to kick off the orchestration and return the status query URI.
|
||||
@app.route(route="singleagent/run", methods=["POST"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def start_single_agent_orchestration(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
"""Start the orchestration and return status metadata."""
|
||||
|
||||
instance_id = await client.start_new(
|
||||
orchestration_function_name="single_agent_orchestration",
|
||||
)
|
||||
|
||||
logger.info("[HTTP] Started orchestration with instance_id: %s", instance_id)
|
||||
|
||||
status_url = _build_status_url(req.url, instance_id, route="singleagent")
|
||||
|
||||
payload = {
|
||||
"message": "Single-agent orchestration started.",
|
||||
"instanceId": instance_id,
|
||||
"statusQueryGetUri": status_url,
|
||||
}
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(payload),
|
||||
status_code=202,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 6. HTTP endpoint to fetch orchestration status using the original instance ID.
|
||||
@app.route(route="singleagent/status/{instanceId}", methods=["GET"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def get_orchestration_status(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
"""Return orchestration runtime status."""
|
||||
|
||||
instance_id = req.route_params.get("instanceId")
|
||||
if not instance_id:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Missing instanceId"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
status = await client.get_status(instance_id)
|
||||
if status is None:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Instance not found"}),
|
||||
status_code=404,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
response_data: dict[str, Any] = {
|
||||
"instanceId": status.instance_id,
|
||||
"runtimeStatus": status.runtime_status.name if status.runtime_status else None,
|
||||
}
|
||||
|
||||
if status.input_ is not None:
|
||||
response_data["input"] = status.input_
|
||||
|
||||
if status.output is not None:
|
||||
response_data["output"] = status.output
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(response_data),
|
||||
status_code=200,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 7. Helper to construct durable status URLs similar to the .NET sample implementation.
|
||||
def _build_status_url(request_url: str, instance_id: str, *, route: str) -> str:
|
||||
"""Construct the status query URI similar to DurableHttpApiExtensions in C#."""
|
||||
|
||||
# Split once on /api/ to preserve host and scheme in local emulator and Azure.
|
||||
base_url, _, _ = request_url.partition("/api/")
|
||||
if not base_url:
|
||||
base_url = request_url.rstrip("/")
|
||||
return f"{base_url}/api/{route}/status/{instance_id}"
|
||||
|
||||
|
||||
"""
|
||||
Expected output when calling `POST /api/singleagent/run` and following the returned status URL:
|
||||
|
||||
HTTP/1.1 202 Accepted
|
||||
{
|
||||
"message": "Single-agent orchestration started.",
|
||||
"instanceId": "<guid>",
|
||||
"statusQueryGetUri": "http://localhost:7071/api/singleagent/status/<guid>"
|
||||
}
|
||||
|
||||
Subsequent `GET /api/singleagent/status/<guid>` after completion returns:
|
||||
|
||||
HTTP/1.1 200 OK
|
||||
{
|
||||
"instanceId": "<guid>",
|
||||
"runtimeStatus": "Completed",
|
||||
"output": "Learning is a journey where curiosity turns effort into mastery."
|
||||
}
|
||||
"""
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"extensionBundle": {
|
||||
"id": "Microsoft.Azure.Functions.ExtensionBundle",
|
||||
"version": "[4.*, 5.0.0)"
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "%TASKHUB_NAME%"
|
||||
}
|
||||
}
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"IsEncrypted": false,
|
||||
"Values": {
|
||||
"FUNCTIONS_WORKER_RUNTIME": "python",
|
||||
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
|
||||
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
|
||||
"TASKHUB_NAME": "default",
|
||||
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
|
||||
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "<AZURE_OPENAI_CHAT_DEPLOYMENT_NAME>",
|
||||
"AZURE_OPENAI_API_KEY": "<AZURE_OPENAI_API_KEY>"
|
||||
}
|
||||
}
|
||||
+2
@@ -0,0 +1,2 @@
|
||||
agent-framework-azurefunctions
|
||||
azure-identity
|
||||
+58
@@ -0,0 +1,58 @@
|
||||
# Multi-Agent Orchestration (Concurrency) – Python
|
||||
|
||||
This sample starts a Durable Functions orchestration that runs two agents in parallel and merges their responses.
|
||||
|
||||
## Highlights
|
||||
- Two agents (`PhysicistAgent` and `ChemistAgent`) share a single Azure OpenAI deployment configuration.
|
||||
- The orchestration uses `context.task_all(...)` to safely run both agents concurrently.
|
||||
- HTTP routes (`/api/multiagent/run` and `/api/multiagent/status/{instanceId}`) mirror the .NET sample for parity.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Use the shared setup instructions in `../README.md` to prepare the environment, install dependencies, and configure Azure OpenAI and storage settings before running this sample.
|
||||
|
||||
## Running the Sample
|
||||
Start the orchestration:
|
||||
|
||||
```bash
|
||||
curl -X POST \
|
||||
-H "Content-Type: text/plain" \
|
||||
--data "What is temperature?" \
|
||||
http://localhost:7071/api/multiagent/run
|
||||
```
|
||||
|
||||
Poll the returned `statusQueryGetUri` until completion:
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/multiagent/status/<instanceId>
|
||||
```
|
||||
|
||||
> **Note:** The agent run endpoints wait for responses by default. If you call them directly and need an immediate HTTP 202, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request payload.
|
||||
|
||||
The orchestration launches both agents simultaneously so their domain-specific answers can be combined for the caller.
|
||||
|
||||
## Expected Output
|
||||
|
||||
Example response when starting the orchestration:
|
||||
|
||||
```json
|
||||
{
|
||||
"message": "Multi-agent concurrent orchestration started.",
|
||||
"prompt": "What is temperature?",
|
||||
"instanceId": "94d56266f0a04e5a8f9f3a1f77a4c597",
|
||||
"statusQueryGetUri": "http://localhost:7071/api/multiagent/status/94d56266f0a04e5a8f9f3a1f77a4c597"
|
||||
}
|
||||
```
|
||||
|
||||
Example completed status payload:
|
||||
|
||||
```json
|
||||
{
|
||||
"instanceId": "94d56266f0a04e5a8f9f3a1f77a4c597",
|
||||
"runtimeStatus": "Completed",
|
||||
"output": {
|
||||
"physicist": "Temperature measures the average kinetic energy of particles in a system.",
|
||||
"chemist": "Temperature reflects how molecular motion influences reaction rates and equilibria."
|
||||
}
|
||||
}
|
||||
```
|
||||
+11
@@ -0,0 +1,11 @@
|
||||
### Start the multi-agent concurrent orchestration
|
||||
POST http://localhost:7071/api/multiagent/run
|
||||
Content-Type: text/plain
|
||||
|
||||
What is temperature?
|
||||
|
||||
### Check the status of the orchestration
|
||||
|
||||
@instanceId =<Enter the instance ID from the response above>
|
||||
|
||||
GET http://localhost:7071/api/multiagent/status/{{instanceId}}
|
||||
+192
@@ -0,0 +1,192 @@
|
||||
"""Fan out concurrent runs across two agents inside a Durable Functions orchestration.
|
||||
|
||||
Components used in this sample:
|
||||
- AzureOpenAIChatClient to create domain-specific agents hosted by Agent Framework.
|
||||
- AgentFunctionApp to expose orchestration and HTTP triggers.
|
||||
- Durable Functions orchestration that executes agent calls in parallel and aggregates results.
|
||||
|
||||
Prerequisites: configure `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME`, and either
|
||||
`AZURE_OPENAI_API_KEY` or authenticate with Azure CLI before starting the Functions host."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import azure.durable_functions as df
|
||||
import azure.functions as func
|
||||
from agent_framework.azure import AgentFunctionApp, AzureOpenAIChatClient
|
||||
from azure.durable_functions import DurableOrchestrationContext
|
||||
from azure.identity import AzureCliCredential
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 1. Define agent names shared across the orchestration.
|
||||
PHYSICIST_AGENT_NAME = "PhysicistAgent"
|
||||
CHEMIST_AGENT_NAME = "ChemistAgent"
|
||||
|
||||
|
||||
# 2. Instantiate both agents that the orchestration will run concurrently.
|
||||
def _create_agents() -> list[Any]:
|
||||
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
|
||||
|
||||
physicist = chat_client.create_agent(
|
||||
name=PHYSICIST_AGENT_NAME,
|
||||
instructions="You are an expert in physics. You answer questions from a physics perspective.",
|
||||
)
|
||||
|
||||
chemist = chat_client.create_agent(
|
||||
name=CHEMIST_AGENT_NAME,
|
||||
instructions="You are an expert in chemistry. You answer questions from a chemistry perspective.",
|
||||
)
|
||||
|
||||
return [physicist, chemist]
|
||||
|
||||
|
||||
# 3. Register both agents with AgentFunctionApp and selectively enable HTTP endpoints.
|
||||
agents = _create_agents()
|
||||
app = AgentFunctionApp(enable_health_check=True, enable_http_endpoints=False)
|
||||
app.add_agent(agents[0], enable_http_endpoint=True)
|
||||
app.add_agent(agents[1])
|
||||
|
||||
|
||||
# 4. Durable Functions orchestration that runs both agents in parallel.
|
||||
@app.orchestration_trigger(context_name="context")
|
||||
def multi_agent_concurrent_orchestration(context: DurableOrchestrationContext):
|
||||
"""Fan out to two domain-specific agents and aggregate their responses."""
|
||||
|
||||
prompt = context.get_input()
|
||||
if not prompt or not str(prompt).strip():
|
||||
raise ValueError("Prompt is required")
|
||||
|
||||
physicist = app.get_agent(context, PHYSICIST_AGENT_NAME)
|
||||
chemist = app.get_agent(context, CHEMIST_AGENT_NAME)
|
||||
|
||||
physicist_thread = physicist.get_new_thread()
|
||||
chemist_thread = chemist.get_new_thread()
|
||||
|
||||
physicist_task = physicist.run(messages=str(prompt), thread=physicist_thread)
|
||||
chemist_task = chemist.run(messages=str(prompt), thread=chemist_thread)
|
||||
|
||||
results = yield context.task_all([physicist_task, chemist_task])
|
||||
|
||||
return {
|
||||
"physicist": results[0].get("response", ""),
|
||||
"chemist": results[1].get("response", ""),
|
||||
}
|
||||
|
||||
|
||||
# 5. HTTP endpoint to accept prompts and start the concurrent orchestration.
|
||||
@app.route(route="multiagent/run", methods=["POST"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def start_multi_agent_concurrent_orchestration(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
"""Kick off the orchestration with a plain text prompt."""
|
||||
|
||||
body_bytes = req.get_body() or b""
|
||||
prompt = body_bytes.decode("utf-8", errors="replace").strip()
|
||||
if not prompt:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Prompt is required"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
instance_id = await client.start_new(
|
||||
orchestration_function_name="multi_agent_concurrent_orchestration",
|
||||
client_input=prompt,
|
||||
)
|
||||
|
||||
logger.info("[HTTP] Started orchestration with instance_id: %s", instance_id)
|
||||
|
||||
status_url = _build_status_url(req.url, instance_id, route="multiagent")
|
||||
|
||||
payload = {
|
||||
"message": "Multi-agent concurrent orchestration started.",
|
||||
"prompt": prompt,
|
||||
"instanceId": instance_id,
|
||||
"statusQueryGetUri": status_url,
|
||||
}
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(payload),
|
||||
status_code=202,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 6. HTTP endpoint to retrieve orchestration status and aggregated outputs.
|
||||
@app.route(route="multiagent/status/{instanceId}", methods=["GET"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def get_orchestration_status(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
instance_id = req.route_params.get("instanceId")
|
||||
if not instance_id:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Missing instanceId"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
status = await client.get_status(instance_id)
|
||||
if status is None:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Instance not found"}),
|
||||
status_code=404,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
response_data: dict[str, Any] = {
|
||||
"instanceId": status.instance_id,
|
||||
"runtimeStatus": status.runtime_status.name if status.runtime_status else None,
|
||||
"createdTime": status.created_time.isoformat() if status.created_time else None,
|
||||
"lastUpdatedTime": status.last_updated_time.isoformat() if status.last_updated_time else None,
|
||||
}
|
||||
|
||||
if status.input_ is not None:
|
||||
response_data["input"] = status.input_
|
||||
|
||||
if status.output is not None:
|
||||
response_data["output"] = status.output
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(response_data),
|
||||
status_code=200,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 7. Helper to construct durable status URLs.
|
||||
def _build_status_url(request_url: str, instance_id: str, *, route: str) -> str:
|
||||
base_url, _, _ = request_url.partition("/api/")
|
||||
if not base_url:
|
||||
base_url = request_url.rstrip("/")
|
||||
return f"{base_url}/api/{route}/status/{instance_id}"
|
||||
|
||||
|
||||
"""
|
||||
Expected output when calling `POST /api/multiagent/run` with a plain-text prompt:
|
||||
|
||||
HTTP/1.1 202 Accepted
|
||||
{
|
||||
"message": "Multi-agent concurrent orchestration started.",
|
||||
"prompt": "What is temperature?",
|
||||
"instanceId": "<guid>",
|
||||
"statusQueryGetUri": "http://localhost:7071/api/multiagent/status/<guid>"
|
||||
}
|
||||
|
||||
Polling `GET /api/multiagent/status/<guid>` after completion returns:
|
||||
|
||||
HTTP/1.1 200 OK
|
||||
{
|
||||
"instanceId": "<guid>",
|
||||
"runtimeStatus": "Completed",
|
||||
"output": {
|
||||
"physicist": "Temperature measures the average kinetic energy of particles in a system.",
|
||||
"chemist": "Temperature reflects how molecular motion influences reaction rates and equilibria."
|
||||
}
|
||||
}
|
||||
"""
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"extensionBundle": {
|
||||
"id": "Microsoft.Azure.Functions.ExtensionBundle",
|
||||
"version": "[4.*, 5.0.0)"
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "%TASKHUB_NAME%"
|
||||
}
|
||||
}
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"IsEncrypted": false,
|
||||
"Values": {
|
||||
"FUNCTIONS_WORKER_RUNTIME": "python",
|
||||
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
|
||||
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
|
||||
"TASKHUB_NAME": "default",
|
||||
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
|
||||
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "<AZURE_OPENAI_CHAT_DEPLOYMENT_NAME>",
|
||||
"AZURE_OPENAI_API_KEY": "<AZURE_OPENAI_API_KEY>"
|
||||
}
|
||||
}
|
||||
+2
@@ -0,0 +1,2 @@
|
||||
agent-framework-azurefunctions
|
||||
azure-identity
|
||||
+35
@@ -0,0 +1,35 @@
|
||||
# Multi-Agent Orchestration (Conditionals) – Python
|
||||
|
||||
This sample evaluates incoming emails with a spam detector agent and,
|
||||
when appropriate, drafts a response using an email assistant agent.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Set up the shared prerequisites outlined in `../README.md`, including the virtual environment, dependency installation, and Azure OpenAI and storage configuration.
|
||||
|
||||
## Scenario Overview
|
||||
- Two Azure OpenAI agents share a single deployment: one flags spam, the other drafts replies.
|
||||
- Structured responses (`is_spam` and `reason`, or `response`) determine which orchestration branch runs.
|
||||
- Activity functions handle the side effects of spam handling and email sending.
|
||||
|
||||
## Running the Sample
|
||||
Submit an email payload:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/spamdetection/run \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"subject": "Sale now on", "body": "Limited time offer"}'
|
||||
```
|
||||
|
||||
Poll the returned `statusQueryGetUri` or call the status route directly:
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/spamdetection/status/<instanceId>
|
||||
```
|
||||
|
||||
> **Note:** The spam detection run endpoint waits for responses by default. To opt into an immediate HTTP 202, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the POST body.
|
||||
|
||||
## Expected Responses
|
||||
- Spam payloads return `Email marked as spam: <reason>` by invoking the `handle_spam_email` activity.
|
||||
- Legitimate emails return `Email sent: <draft>` after the email assistant agent produces a structured reply.
|
||||
- The status endpoint mirrors Durable Functions metadata, including runtime status and the agent output.
|
||||
+24
@@ -0,0 +1,24 @@
|
||||
### Test spam detection with a legitimate email
|
||||
POST http://localhost:7071/api/spamdetection/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"email_id": "email-001",
|
||||
"email_content": "Hi John, I hope you're doing well. I wanted to follow up on our meeting yesterday about the quarterly report. Could you please send me the updated figures by Friday? Thanks!"
|
||||
}
|
||||
|
||||
|
||||
### Test spam detection with a spam email
|
||||
POST http://localhost:7071/api/spamdetection/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"email_id": "email-002",
|
||||
"email_content": "URGENT! You've won $1,000,000! Click here now to claim your prize! Limited time offer! Don't miss out!"
|
||||
}
|
||||
|
||||
|
||||
### Check the status of the orchestration
|
||||
@instanceId =<Replace with the instance ID from the response above>
|
||||
|
||||
GET http://localhost:7071/api/spamdetection/status/{{instanceId}}
|
||||
+273
@@ -0,0 +1,273 @@
|
||||
"""Route email requests through conditional orchestration with two agents.
|
||||
|
||||
Components used in this sample:
|
||||
- AzureOpenAIChatClient agents for spam detection and email drafting.
|
||||
- AgentFunctionApp with Durable orchestration, activity, and HTTP triggers.
|
||||
- Pydantic models that validate payloads and agent JSON responses.
|
||||
|
||||
Prerequisites: set `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME`,
|
||||
and either `AZURE_OPENAI_API_KEY` or sign in with Azure CLI before running the
|
||||
Functions host."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, cast
|
||||
|
||||
import azure.durable_functions as df
|
||||
import azure.functions as func
|
||||
from agent_framework.azure import AgentFunctionApp, AzureOpenAIChatClient
|
||||
from azure.durable_functions import DurableOrchestrationContext
|
||||
from azure.identity import AzureCliCredential
|
||||
from pydantic import BaseModel, ValidationError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 1. Define agent names shared across the orchestration.
|
||||
SPAM_AGENT_NAME = "SpamDetectionAgent"
|
||||
EMAIL_AGENT_NAME = "EmailAssistantAgent"
|
||||
|
||||
|
||||
class SpamDetectionResult(BaseModel):
|
||||
is_spam: bool
|
||||
reason: str
|
||||
|
||||
|
||||
class EmailResponse(BaseModel):
|
||||
response: str
|
||||
|
||||
|
||||
class EmailPayload(BaseModel):
|
||||
email_id: str
|
||||
email_content: str
|
||||
|
||||
# 2. Instantiate both agents so they can be registered with AgentFunctionApp.
|
||||
def _create_agents() -> list[Any]:
|
||||
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
|
||||
|
||||
spam_agent = chat_client.create_agent(
|
||||
name=SPAM_AGENT_NAME,
|
||||
instructions="You are a spam detection assistant that identifies spam emails.",
|
||||
)
|
||||
|
||||
email_agent = chat_client.create_agent(
|
||||
name=EMAIL_AGENT_NAME,
|
||||
instructions="You are an email assistant that helps users draft responses to emails with professionalism.",
|
||||
)
|
||||
|
||||
return [spam_agent, email_agent]
|
||||
|
||||
|
||||
app = AgentFunctionApp(agents=_create_agents(), enable_health_check=True)
|
||||
|
||||
|
||||
# 3. Activities handle the side effects for spam and legitimate emails.
|
||||
@app.activity_trigger(input_name="reason")
|
||||
def handle_spam_email(reason: str) -> str:
|
||||
return f"Email marked as spam: {reason}"
|
||||
|
||||
|
||||
@app.activity_trigger(input_name="message")
|
||||
def send_email(message: str) -> str:
|
||||
return f"Email sent: {message}"
|
||||
|
||||
|
||||
# 4. Orchestration validates input, runs agents, and branches on spam results.
|
||||
@app.orchestration_trigger(context_name="context")
|
||||
def spam_detection_orchestration(context: DurableOrchestrationContext):
|
||||
payload_raw = context.get_input()
|
||||
if not isinstance(payload_raw, Mapping):
|
||||
raise ValueError("Email data is required")
|
||||
|
||||
try:
|
||||
payload = EmailPayload.model_validate(payload_raw)
|
||||
except ValidationError as exc:
|
||||
raise ValueError(f"Invalid email payload: {exc}") from exc
|
||||
|
||||
spam_agent = app.get_agent(context, SPAM_AGENT_NAME)
|
||||
email_agent = app.get_agent(context, EMAIL_AGENT_NAME)
|
||||
|
||||
spam_thread = spam_agent.get_new_thread()
|
||||
|
||||
spam_prompt = (
|
||||
"Analyze this email for spam content and return a JSON response with 'is_spam' (boolean) "
|
||||
"and 'reason' (string) fields:\n"
|
||||
f"Email ID: {payload.email_id}\n"
|
||||
f"Content: {payload.email_content}"
|
||||
)
|
||||
|
||||
spam_result_raw = yield spam_agent.run(
|
||||
messages=spam_prompt,
|
||||
thread=spam_thread,
|
||||
response_format=SpamDetectionResult,
|
||||
)
|
||||
|
||||
spam_result = cast(SpamDetectionResult, _coerce_structured(spam_result_raw, SpamDetectionResult))
|
||||
|
||||
if spam_result.is_spam:
|
||||
result = yield context.call_activity("handle_spam_email", spam_result.reason)
|
||||
return result
|
||||
|
||||
email_thread = email_agent.get_new_thread()
|
||||
|
||||
email_prompt = (
|
||||
"Draft a professional response to this email. Return a JSON response with a 'response' field "
|
||||
"containing the reply:\n\n"
|
||||
f"Email ID: {payload.email_id}\n"
|
||||
f"Content: {payload.email_content}"
|
||||
)
|
||||
|
||||
email_result_raw = yield email_agent.run(
|
||||
messages=email_prompt,
|
||||
thread=email_thread,
|
||||
response_format=EmailResponse,
|
||||
)
|
||||
|
||||
email_result = cast(EmailResponse, _coerce_structured(email_result_raw, EmailResponse))
|
||||
|
||||
result = yield context.call_activity("send_email", email_result.response)
|
||||
return result
|
||||
|
||||
|
||||
# 5. HTTP starter endpoint launches the orchestration for each email payload.
|
||||
@app.route(route="spamdetection/run", methods=["POST"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def start_spam_detection_orchestration(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
try:
|
||||
body = req.get_json()
|
||||
except ValueError:
|
||||
body = None
|
||||
|
||||
if not isinstance(body, Mapping):
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Email data is required"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
try:
|
||||
payload = EmailPayload.model_validate(body)
|
||||
except ValidationError as exc:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": f"Invalid email payload: {exc}"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
instance_id = await client.start_new(
|
||||
orchestration_function_name="spam_detection_orchestration",
|
||||
client_input=payload.model_dump(),
|
||||
)
|
||||
|
||||
logger.info("[HTTP] Started spam detection orchestration with instance_id: %s", instance_id)
|
||||
|
||||
status_url = _build_status_url(req.url, instance_id, route="spamdetection")
|
||||
|
||||
payload_json = {
|
||||
"message": "Spam detection orchestration started.",
|
||||
"emailId": payload.email_id,
|
||||
"instanceId": instance_id,
|
||||
"statusQueryGetUri": status_url,
|
||||
}
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(payload_json),
|
||||
status_code=202,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 6. Status endpoint mirrors Durable Functions default payload with agent data.
|
||||
@app.route(route="spamdetection/status/{instanceId}", methods=["GET"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def get_orchestration_status(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
instance_id = req.route_params.get("instanceId")
|
||||
if not instance_id:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Missing instanceId"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
status = await client.get_status(instance_id)
|
||||
if status is None:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Instance not found"}),
|
||||
status_code=404,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
response_data: dict[str, Any] = {
|
||||
"instanceId": status.instance_id,
|
||||
"runtimeStatus": status.runtime_status.name if status.runtime_status else None,
|
||||
"createdTime": status.created_time.isoformat() if status.created_time else None,
|
||||
"lastUpdatedTime": status.last_updated_time.isoformat() if status.last_updated_time else None,
|
||||
}
|
||||
|
||||
if status.input_ is not None:
|
||||
response_data["input"] = status.input_
|
||||
|
||||
if status.output is not None:
|
||||
response_data["output"] = status.output
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(response_data),
|
||||
status_code=200,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 7. Helper utilities keep URL construction and structured parsing deterministic.
|
||||
def _build_status_url(request_url: str, instance_id: str, *, route: str) -> str:
|
||||
base_url, _, _ = request_url.partition("/api/")
|
||||
if not base_url:
|
||||
base_url = request_url.rstrip("/")
|
||||
return f"{base_url}/api/{route}/status/{instance_id}"
|
||||
|
||||
|
||||
def _coerce_structured(result: Mapping[str, Any], model: type[BaseModel]) -> BaseModel:
|
||||
structured = result.get("structured_response") if isinstance(result, Mapping) else None
|
||||
if structured is not None:
|
||||
return model.model_validate(structured)
|
||||
|
||||
response_text = result.get("response") if isinstance(result, Mapping) else None
|
||||
if isinstance(response_text, str) and response_text.strip():
|
||||
try:
|
||||
parsed = json.loads(response_text)
|
||||
if isinstance(parsed, Mapping):
|
||||
return model.model_validate(parsed)
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("[ConditionalOrchestration] Failed to parse agent JSON response; raising error.")
|
||||
|
||||
# If parsing failed, raise to surface the issue to the caller.
|
||||
raise ValueError(f"Agent response could not be parsed as {model.__name__}.")
|
||||
|
||||
|
||||
"""
|
||||
Expected response from `POST /api/spamdetection/run`:
|
||||
|
||||
HTTP/1.1 202 Accepted
|
||||
{
|
||||
"message": "Spam detection orchestration started.",
|
||||
"emailId": "123",
|
||||
"instanceId": "<durable-instance-id>",
|
||||
"statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/<durable-instance-id>"
|
||||
}
|
||||
|
||||
Expected response from `GET /api/spamdetection/status/{instanceId}` once complete:
|
||||
|
||||
HTTP/1.1 200 OK
|
||||
{
|
||||
"instanceId": "<durable-instance-id>",
|
||||
"runtimeStatus": "Completed",
|
||||
"createdTime": "2024-01-01T00:00:00+00:00",
|
||||
"lastUpdatedTime": "2024-01-01T00:00:10+00:00",
|
||||
"output": "Email sent: Thank you for reaching out..."
|
||||
}
|
||||
"""
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"extensionBundle": {
|
||||
"id": "Microsoft.Azure.Functions.ExtensionBundle",
|
||||
"version": "[4.*, 5.0.0)"
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "%TASKHUB_NAME%"
|
||||
}
|
||||
}
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"IsEncrypted": false,
|
||||
"Values": {
|
||||
"FUNCTIONS_WORKER_RUNTIME": "python",
|
||||
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
|
||||
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
|
||||
"TASKHUB_NAME": "default",
|
||||
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
|
||||
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "<AZURE_OPENAI_CHAT_DEPLOYMENT_NAME>",
|
||||
"AZURE_OPENAI_API_KEY": "<AZURE_OPENAI_API_KEY>"
|
||||
}
|
||||
}
|
||||
+2
@@ -0,0 +1,2 @@
|
||||
agent-framework-azurefunctions
|
||||
azure-identity
|
||||
+48
@@ -0,0 +1,48 @@
|
||||
# Single-Agent Orchestration (HITL) – Python
|
||||
|
||||
This sample demonstrates the human-in-the-loop (HITL) scenario.
|
||||
A single writer agent iterates on content until a human reviewer approves the
|
||||
output or a maximum number of attempts is reached.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Complete the common setup instructions in `../README.md` to prepare the virtual environment, install dependencies, and configure Azure OpenAI and storage settings.
|
||||
|
||||
## What It Shows
|
||||
- Identical environment variable usage (`AZURE_OPENAI_ENDPOINT`,
|
||||
`AZURE_OPENAI_DEPLOYMENT`) and HTTP surface area (`/api/hitl/...`).
|
||||
- Durable orchestrations that pause for external events while maintaining
|
||||
deterministic state (`context.wait_for_external_event` + timed cancellation).
|
||||
- Activity functions that encapsulate the out-of-band operations such as notifying
|
||||
a reviewer and publishing content.
|
||||
|
||||
## Running the Sample
|
||||
Start the HITL orchestration:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/hitl/run \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"topic": "Write a friendly release note"}'
|
||||
```
|
||||
|
||||
Poll the returned `statusQueryGetUri` or call the status route directly:
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/hitl/status/<instanceId>
|
||||
```
|
||||
|
||||
Approve or reject the draft:
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/hitl/approve/<instanceId> \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"approved": true, "feedback": "Looks good"}'
|
||||
```
|
||||
|
||||
> **Note:** Calls to the underlying agent run endpoint wait for responses by default. If you need an immediate HTTP 202 response, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body.
|
||||
|
||||
## Expected Responses
|
||||
- `POST /api/hitl/run` returns a 202 Accepted payload with the Durable Functions instance ID.
|
||||
- `POST /api/hitl/approve/{instanceId}` echoes the decision that the orchestration receives.
|
||||
- `GET /api/hitl/status/{instanceId}` reports `runtimeStatus`, custom status messages, and the final content when approved.
|
||||
The orchestration sets custom status messages, retries on rejection with reviewer feedback, and raises a timeout if human approval does not arrive.
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
### Start the HITL content generation orchestration with default timeout (72 hours)
|
||||
POST http://localhost:7071/api/hitl/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"topic": "The Future of Artificial Intelligence",
|
||||
"max_review_attempts": 3
|
||||
}
|
||||
|
||||
|
||||
### Start the HITL content generation orchestration with a short timeout (~4 seconds)
|
||||
POST http://localhost:7071/api/hitl/run
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"topic": "The Future of Artificial Intelligence",
|
||||
"max_review_attempts": 3,
|
||||
"approval_timeout_hours": 0.001
|
||||
}
|
||||
|
||||
|
||||
### Replace INSTANCE_ID_GOES_HERE below with the value returned from the POST call
|
||||
@instanceId=<INSTANCE_ID_GOES_HERE>
|
||||
|
||||
### Check the status of the orchestration
|
||||
GET http://localhost:7071/api/hitl/status/{{instanceId}}
|
||||
|
||||
### Send human approval
|
||||
POST http://localhost:7071/api/hitl/approve/{{instanceId}}
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"approved": true,
|
||||
"feedback": "Great article! The content is well-structured and informative."
|
||||
}
|
||||
|
||||
### Send human rejection with feedback
|
||||
POST http://localhost:7071/api/hitl/approve/{{instanceId}}
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"approved": false,
|
||||
"feedback": "The article needs more technical depth and better examples."
|
||||
}
|
||||
|
||||
+399
@@ -0,0 +1,399 @@
|
||||
"""Iterate on generated content with a human-in-the-loop Durable orchestration.
|
||||
|
||||
Components used in this sample:
|
||||
- AzureOpenAIChatClient for a single writer agent that emits structured JSON.
|
||||
- AgentFunctionApp with Durable orchestration, HTTP triggers, and activity triggers.
|
||||
- External events that pause the workflow until a human decision arrives or times out.
|
||||
|
||||
Prerequisites: configure `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME`, and
|
||||
either `AZURE_OPENAI_API_KEY` or sign in with Azure CLI before running `func start`."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Mapping
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
|
||||
import azure.durable_functions as df
|
||||
import azure.functions as func
|
||||
from agent_framework.azure import AgentFunctionApp, AzureOpenAIChatClient
|
||||
from azure.durable_functions import DurableOrchestrationContext
|
||||
from azure.identity import AzureCliCredential
|
||||
from pydantic import BaseModel, ValidationError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 1. Define orchestration constants used throughout the workflow.
|
||||
WRITER_AGENT_NAME = "WriterAgent"
|
||||
HUMAN_APPROVAL_EVENT = "HumanApproval"
|
||||
|
||||
|
||||
class ContentGenerationInput(BaseModel):
|
||||
topic: str
|
||||
max_review_attempts: int = 3
|
||||
approval_timeout_hours: float = 72
|
||||
|
||||
|
||||
class GeneratedContent(BaseModel):
|
||||
title: str
|
||||
content: str
|
||||
|
||||
|
||||
class HumanApproval(BaseModel):
|
||||
approved: bool
|
||||
feedback: str = ""
|
||||
|
||||
|
||||
# 2. Create the writer agent that produces structured JSON responses.
|
||||
def _create_writer_agent() -> Any:
|
||||
instructions = (
|
||||
"You are a professional content writer who creates high-quality articles on various topics. "
|
||||
"You write engaging, informative, and well-structured content that follows best practices for readability and accuracy. "
|
||||
"Return your response as JSON with 'title' and 'content' fields."
|
||||
)
|
||||
|
||||
return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
|
||||
name=WRITER_AGENT_NAME,
|
||||
instructions=instructions,
|
||||
)
|
||||
|
||||
|
||||
app = AgentFunctionApp(agents=[_create_writer_agent()], enable_health_check=True)
|
||||
|
||||
|
||||
# 3. Activities encapsulate external work for review notifications and publishing.
|
||||
@app.activity_trigger(input_name="content")
|
||||
def notify_user_for_approval(content: dict) -> None:
|
||||
model = GeneratedContent.model_validate(content)
|
||||
logger.info("NOTIFICATION: Please review the following content for approval:")
|
||||
logger.info("Title: %s", model.title or "(untitled)")
|
||||
logger.info("Content: %s", model.content)
|
||||
logger.info("Use the approval endpoint to approve or reject this content.")
|
||||
|
||||
|
||||
@app.activity_trigger(input_name="content")
|
||||
def publish_content(content: dict) -> None:
|
||||
model = GeneratedContent.model_validate(content)
|
||||
logger.info("PUBLISHING: Content has been published successfully:")
|
||||
logger.info("Title: %s", model.title or "(untitled)")
|
||||
logger.info("Content: %s", model.content)
|
||||
|
||||
|
||||
# 4. Orchestration loops until the human approves, times out, or attempts are exhausted.
|
||||
@app.orchestration_trigger(context_name="context")
|
||||
def content_generation_hitl_orchestration(context: DurableOrchestrationContext):
|
||||
payload_raw = context.get_input()
|
||||
if not isinstance(payload_raw, Mapping):
|
||||
raise ValueError("Content generation input is required")
|
||||
|
||||
try:
|
||||
payload = ContentGenerationInput.model_validate(payload_raw)
|
||||
except ValidationError as exc:
|
||||
raise ValueError(f"Invalid content generation input: {exc}") from exc
|
||||
|
||||
writer = app.get_agent(context, WRITER_AGENT_NAME)
|
||||
writer_thread = writer.get_new_thread()
|
||||
|
||||
context.set_custom_status(f"Starting content generation for topic: {payload.topic}")
|
||||
|
||||
initial_raw = yield writer.run(
|
||||
messages=f"Write a short article about '{payload.topic}'.",
|
||||
thread=writer_thread,
|
||||
response_format=GeneratedContent,
|
||||
)
|
||||
content = _coerce_generated_content(initial_raw)
|
||||
|
||||
attempt = 0
|
||||
while attempt < payload.max_review_attempts:
|
||||
attempt += 1
|
||||
context.set_custom_status(
|
||||
f"Requesting human feedback. Iteration #{attempt}. Timeout: {payload.approval_timeout_hours} hour(s)."
|
||||
)
|
||||
|
||||
yield context.call_activity("notify_user_for_approval", content.model_dump())
|
||||
|
||||
approval_task = context.wait_for_external_event(HUMAN_APPROVAL_EVENT)
|
||||
timeout_task = context.create_timer(
|
||||
context.current_utc_datetime + timedelta(hours=payload.approval_timeout_hours)
|
||||
)
|
||||
|
||||
winner = yield context.task_any([approval_task, timeout_task])
|
||||
|
||||
if winner == approval_task:
|
||||
timeout_task.cancel() # type: ignore[attr-defined]
|
||||
approval_payload = _parse_human_approval(approval_task.result)
|
||||
|
||||
if approval_payload.approved:
|
||||
context.set_custom_status("Content approved by human reviewer. Publishing content...")
|
||||
yield context.call_activity("publish_content", content.model_dump())
|
||||
context.set_custom_status(
|
||||
f"Content published successfully at {context.current_utc_datetime:%Y-%m-%dT%H:%M:%S}"
|
||||
)
|
||||
return {"content": content.content}
|
||||
|
||||
context.set_custom_status(
|
||||
"Content rejected by human reviewer. Incorporating feedback and regenerating..."
|
||||
)
|
||||
rewrite_prompt = (
|
||||
"The content was rejected by a human reviewer. Please rewrite the article incorporating their feedback.\n\n"
|
||||
f"Human Feedback: {approval_payload.feedback or 'No feedback provided.'}"
|
||||
)
|
||||
rewritten_raw = yield writer.run(
|
||||
messages=rewrite_prompt,
|
||||
thread=writer_thread,
|
||||
response_format=GeneratedContent,
|
||||
)
|
||||
content = _coerce_generated_content(rewritten_raw)
|
||||
else:
|
||||
context.set_custom_status(
|
||||
f"Human approval timed out after {payload.approval_timeout_hours} hour(s). Treating as rejection."
|
||||
)
|
||||
raise TimeoutError(
|
||||
f"Human approval timed out after {payload.approval_timeout_hours} hour(s)."
|
||||
)
|
||||
|
||||
raise RuntimeError(f"Content could not be approved after {payload.max_review_attempts} iteration(s).")
|
||||
|
||||
|
||||
# 5. HTTP endpoint that starts the human-in-the-loop orchestration.
|
||||
@app.route(route="hitl/run", methods=["POST"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def start_content_generation(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
try:
|
||||
body = req.get_json()
|
||||
except ValueError:
|
||||
body = None
|
||||
|
||||
if not isinstance(body, Mapping):
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Request body must be valid JSON."}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
try:
|
||||
payload = ContentGenerationInput.model_validate(body)
|
||||
except ValidationError as exc:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": f"Invalid content generation input: {exc}"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
instance_id = await client.start_new(
|
||||
orchestration_function_name="content_generation_hitl_orchestration",
|
||||
client_input=payload.model_dump(),
|
||||
)
|
||||
|
||||
status_url = _build_status_url(req.url, instance_id, route="hitl")
|
||||
|
||||
payload_json = {
|
||||
"message": "HITL content generation orchestration started.",
|
||||
"topic": payload.topic,
|
||||
"instanceId": instance_id,
|
||||
"statusQueryGetUri": status_url,
|
||||
}
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(payload_json),
|
||||
status_code=202,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 6. Endpoint that delivers human approval or rejection back into the orchestration.
|
||||
@app.route(route="hitl/approve/{instanceId}", methods=["POST"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def send_human_approval(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
instance_id = req.route_params.get("instanceId")
|
||||
if not instance_id:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Missing instanceId in route."}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
try:
|
||||
body = req.get_json()
|
||||
except ValueError:
|
||||
body = None
|
||||
|
||||
if not isinstance(body, Mapping):
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Approval response is required"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
try:
|
||||
approval = HumanApproval.model_validate(body)
|
||||
except ValidationError as exc:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": f"Invalid approval payload: {exc}"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
await client.raise_event(instance_id, HUMAN_APPROVAL_EVENT, approval.model_dump())
|
||||
|
||||
payload_json = {
|
||||
"message": "Human approval sent to orchestration.",
|
||||
"instanceId": instance_id,
|
||||
"approved": approval.approved,
|
||||
}
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(payload_json),
|
||||
status_code=200,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 7. Endpoint that mirrors Durable Functions status plus custom workflow messaging.
|
||||
@app.route(route="hitl/status/{instanceId}", methods=["GET"])
|
||||
@app.durable_client_input(client_name="client")
|
||||
async def get_orchestration_status(
|
||||
req: func.HttpRequest,
|
||||
client: df.DurableOrchestrationClient,
|
||||
) -> func.HttpResponse:
|
||||
instance_id = req.route_params.get("instanceId")
|
||||
if not instance_id:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Missing instanceId"}),
|
||||
status_code=400,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
status = await client.get_status(
|
||||
instance_id,
|
||||
show_history=False,
|
||||
show_history_output=False,
|
||||
show_input=True,
|
||||
)
|
||||
|
||||
# Check if status is None or if the instance doesn't exist (runtime_status is None)
|
||||
if status is None or getattr(status, "runtime_status", None) is None:
|
||||
return func.HttpResponse(
|
||||
body=json.dumps({"error": "Instance not found."}),
|
||||
status_code=404,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
response_data: dict[str, Any] = {
|
||||
"instanceId": getattr(status, "instance_id", None),
|
||||
"runtimeStatus": getattr(status.runtime_status, "name", None)
|
||||
if getattr(status, "runtime_status", None)
|
||||
else None,
|
||||
"workflowStatus": getattr(status, "custom_status", None),
|
||||
}
|
||||
|
||||
if getattr(status, "input_", None) is not None:
|
||||
response_data["input"] = status.input_
|
||||
|
||||
if getattr(status, "output", None) is not None:
|
||||
response_data["output"] = status.output
|
||||
|
||||
failure_details = getattr(status, "failure_details", None)
|
||||
if failure_details is not None:
|
||||
response_data["failureDetails"] = failure_details
|
||||
|
||||
return func.HttpResponse(
|
||||
body=json.dumps(response_data),
|
||||
status_code=200,
|
||||
mimetype="application/json",
|
||||
)
|
||||
|
||||
|
||||
# 8. Helper utilities keep parsing logic deterministic.
|
||||
def _build_status_url(request_url: str, instance_id: str, *, route: str) -> str:
|
||||
base_url, _, _ = request_url.partition("/api/")
|
||||
if not base_url:
|
||||
base_url = request_url.rstrip("/")
|
||||
return f"{base_url}/api/{route}/status/{instance_id}"
|
||||
|
||||
|
||||
def _coerce_generated_content(result: Mapping[str, Any]) -> GeneratedContent:
|
||||
structured = result.get("structured_response") if isinstance(result, Mapping) else None
|
||||
if structured is not None:
|
||||
return GeneratedContent.model_validate(structured)
|
||||
|
||||
response_text = result.get("response") if isinstance(result, Mapping) else None
|
||||
if isinstance(response_text, str) and response_text.strip():
|
||||
try:
|
||||
parsed = json.loads(response_text)
|
||||
if isinstance(parsed, Mapping):
|
||||
return GeneratedContent.model_validate(parsed)
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("[HITL] Failed to parse agent JSON response; falling back to defaults.")
|
||||
|
||||
raise ValueError("Agent response could not be parsed as GeneratedContent.")
|
||||
|
||||
|
||||
def _parse_human_approval(raw: Any) -> HumanApproval:
|
||||
if isinstance(raw, Mapping):
|
||||
return HumanApproval.model_validate(raw)
|
||||
|
||||
if isinstance(raw, str):
|
||||
stripped = raw.strip()
|
||||
if not stripped:
|
||||
return HumanApproval(approved=False, feedback="")
|
||||
try:
|
||||
parsed = json.loads(stripped)
|
||||
if isinstance(parsed, Mapping):
|
||||
return HumanApproval.model_validate(parsed)
|
||||
except json.JSONDecodeError:
|
||||
logger.debug(
|
||||
"[HITL] Approval payload is not valid JSON; using string heuristics.",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
affirmative = {"true", "yes", "approved", "y", "1"}
|
||||
negative = {"false", "no", "rejected", "n", "0"}
|
||||
lower = stripped.lower()
|
||||
if lower in affirmative:
|
||||
return HumanApproval(approved=True, feedback="")
|
||||
if lower in negative:
|
||||
return HumanApproval(approved=False, feedback="")
|
||||
return HumanApproval(approved=False, feedback=stripped)
|
||||
|
||||
raise ValueError("Approval payload must be a JSON object or string.")
|
||||
|
||||
|
||||
"""
|
||||
Expected response from `POST /api/hitl/run`:
|
||||
|
||||
HTTP/1.1 202 Accepted
|
||||
{
|
||||
"message": "HITL content generation orchestration started.",
|
||||
"topic": "Contoso launch",
|
||||
"instanceId": "<durable-instance-id>",
|
||||
"statusQueryGetUri": "http://localhost:7071/api/hitl/status/<durable-instance-id>"
|
||||
}
|
||||
|
||||
Expected response after approving via `POST /api/hitl/approve/{instanceId}`:
|
||||
|
||||
HTTP/1.1 200 OK
|
||||
{
|
||||
"message": "Human approval sent to orchestration.",
|
||||
"instanceId": "<durable-instance-id>",
|
||||
"approved": true
|
||||
}
|
||||
|
||||
Expected response from `GET /api/hitl/status/{instanceId}` once published:
|
||||
|
||||
HTTP/1.1 200 OK
|
||||
{
|
||||
"instanceId": "<durable-instance-id>",
|
||||
"runtimeStatus": "Completed",
|
||||
"workflowStatus": "Content published successfully at 2024-01-01T12:00:00",
|
||||
"output": {
|
||||
"content": "Thank you for joining the Contoso product launch..."
|
||||
}
|
||||
}
|
||||
"""
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"extensionBundle": {
|
||||
"id": "Microsoft.Azure.Functions.ExtensionBundle",
|
||||
"version": "[4.*, 5.0.0)"
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "%TASKHUB_NAME%"
|
||||
}
|
||||
}
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"IsEncrypted": false,
|
||||
"Values": {
|
||||
"FUNCTIONS_WORKER_RUNTIME": "python",
|
||||
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
|
||||
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
|
||||
"TASKHUB_NAME": "default",
|
||||
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
|
||||
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "<AZURE_OPENAI_CHAT_DEPLOYMENT_NAME>",
|
||||
"AZURE_OPENAI_API_KEY": "<AZURE_OPENAI_API_KEY>"
|
||||
}
|
||||
}
|
||||
+2
@@ -0,0 +1,2 @@
|
||||
agent-framework-azurefunctions
|
||||
azure-identity
|
||||
@@ -0,0 +1,48 @@
|
||||
These are common instructions for setting up your environment for every sample in this directory.
|
||||
These samples illustrate the Durable extensibility for Agent Framework running in Azure Functions.
|
||||
|
||||
All of these samples are set up to run in Azure Functions. Azure Functions has a local development tool called [CoreTools](https://learn.microsoft.com/azure/azure-functions/functions-run-local?tabs=windows%2Cpython%2Cv2&pivots=programming-language-python#install-the-azure-functions-core-tools) which we will set up to run these samples locally.
|
||||
|
||||
## Environment Setup
|
||||
|
||||
### 1. Install dependencies and create appropriate services
|
||||
|
||||
- Install [Azure Functions Core Tools 4.x](https://learn.microsoft.com/azure/azure-functions/functions-run-local?tabs=windows%2Cpython%2Cv2&pivots=programming-language-python#install-the-azure-functions-core-tools)
|
||||
|
||||
- Install [Azurite storage emulator](https://learn.microsoft.com/en-us/azure/storage/common/storage-install-azurite?toc=%2Fazure%2Fstorage%2Fblobs%2Ftoc.json&bc=%2Fazure%2Fstorage%2Fblobs%2Fbreadcrumb%2Ftoc.json&tabs=visual-studio%2Cblob-storage)
|
||||
|
||||
- Create an [Azure OpenAI](https://azure.microsoft.com/en-us/products/ai-foundry/models/openai) resource. Note the Azure OpenAI endpoint, deployment name, and the key (or ensure you can authenticate with `AzureCliCredential`).
|
||||
|
||||
- Install a tool to execute HTTP calls, for example the [REST Client extension](https://marketplace.visualstudio.com/items?itemName=humao.rest-client)
|
||||
|
||||
- [Optionally] Create an [Azure Function Python app](https://learn.microsoft.com/en-us/azure/azure-functions/functions-create-function-app-portal?tabs=core-tools&pivots=flex-consumption-plan) to later deploy your app to Azure if you so desire.
|
||||
|
||||
### 2. Create and activate a virtual environment
|
||||
|
||||
**Windows (PowerShell):**
|
||||
```powershell
|
||||
python -m venv .venv
|
||||
.venv\Scripts\Activate.ps1
|
||||
```
|
||||
|
||||
**Linux/macOS:**
|
||||
```bash
|
||||
python -m venv .venv
|
||||
source .venv/bin/activate
|
||||
```
|
||||
|
||||
### 3. Running the samples
|
||||
|
||||
- [Start the Azurite emulator](https://learn.microsoft.com/en-us/azure/storage/common/storage-install-azurite?tabs=npm%2Cblob-storage#run-azurite)
|
||||
|
||||
- Inside each sample:
|
||||
|
||||
- Install Python dependencies – from the sample directory, run `pip install -r requirements.txt` (or the equivalent in your active virtual environment).
|
||||
|
||||
- Copy `local.settings.json.template` to `local.settings.json`, then update `AZURE_OPENAI_ENDPOINT` and `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME` for Azure OpenAI authentication. The samples use `AzureCliCredential` by default, so ensure you're logged in via `az login`.
|
||||
- Alternatively, you can use API key authentication by setting `AZURE_OPENAI_API_KEY` and updating the code to use `AzureOpenAIChatClient()` without the credential parameter.
|
||||
- Keep `TASKHUB_NAME` set to `default` unless you plan to change the durable task hub name.
|
||||
|
||||
- Run the command `func start` from the root of the sample
|
||||
|
||||
- Follow each sample's README for scenario-specific steps, and use its `demo.http` file (or provided curl examples) to trigger the hosted HTTP endpoints.
|
||||
Reference in New Issue
Block a user