mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Add DevUI to AgentFramework (#781)
* add initial backend service code for devui * add tests * add frontendcode * ui updates * update readme * ui updates and tweaks * update ui bundle * improve ui, add react flow base * add react flow ui, fix background * update ui, fix introspection bug * update readme * update ui build * add support for multimodal input - both backend and frontend * update ui build * refactor as main framework package * backend and tests refactor * ui build update * ui build update and refactor * update pyproject.toml, update uv.lock * update ui build * ui update to fit oai responses types * add backend updat and readme update * mypy and other fixes * add intial dev guide * update ui and fix workflow bug * update ui build, add thread support * type fixes * update workflow view * update uv.lock * fix workflow iport errors * lint and other fixes * mypy fixes * minor update * update ui build * refactor to use oai dependencies directly, update examples to samples, improve typing * readme update * update ui and ui build * fix workflow pyright error * update ui, fix issues with run workflow placement, miniamp menu, etc * make samples integrate serve --------- Co-authored-by: Chris <66376200+crickman@users.noreply.github.com> Co-authored-by: Eric Zhu <ekzhu@users.noreply.github.com>
This commit is contained in:
Victor Dibia
committed by
GitHub
Unverified
GitHub
parent
adb6dcd2af
commit
1ef24d3e91
@@ -0,0 +1,3 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Examples package for Agent Framework DevUI."""
|
||||
@@ -0,0 +1,3 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Fanout workflow example."""
|
||||
@@ -0,0 +1,698 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Complex Fan-In/Fan-Out Data Processing Workflow.
|
||||
|
||||
This workflow demonstrates a sophisticated data processing pipeline with multiple stages:
|
||||
1. Data Ingestion - Simulates loading data from multiple sources
|
||||
2. Data Validation - Multiple validators run in parallel to check data quality
|
||||
3. Data Transformation - Fan-out to different transformation processors
|
||||
4. Quality Assurance - Multiple QA checks run in parallel
|
||||
5. Data Aggregation - Fan-in to combine processed results
|
||||
6. Final Processing - Generate reports and complete workflow
|
||||
|
||||
The workflow includes realistic delays to simulate actual processing time and
|
||||
shows complex fan-in/fan-out patterns with conditional processing.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Literal
|
||||
|
||||
from agent_framework import (
|
||||
Executor,
|
||||
WorkflowBuilder,
|
||||
WorkflowCompletedEvent,
|
||||
WorkflowContext,
|
||||
handler,
|
||||
)
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class DataType(Enum):
|
||||
"""Types of data being processed."""
|
||||
|
||||
CUSTOMER = "customer"
|
||||
TRANSACTION = "transaction"
|
||||
PRODUCT = "product"
|
||||
ANALYTICS = "analytics"
|
||||
|
||||
|
||||
class ValidationResult(Enum):
|
||||
"""Results of data validation."""
|
||||
|
||||
VALID = "valid"
|
||||
WARNING = "warning"
|
||||
ERROR = "error"
|
||||
|
||||
|
||||
class ProcessingRequest(BaseModel):
|
||||
"""Complex input structure for data processing workflow."""
|
||||
|
||||
# Basic information
|
||||
data_source: Literal["database", "api", "file_upload", "streaming"] = Field(
|
||||
description="The source of the data to be processed", default="database"
|
||||
)
|
||||
|
||||
data_type: Literal["customer", "transaction", "product", "analytics"] = Field(
|
||||
description="Type of data being processed", default="customer"
|
||||
)
|
||||
|
||||
processing_priority: Literal["low", "normal", "high", "critical"] = Field(
|
||||
description="Processing priority level", default="normal"
|
||||
)
|
||||
|
||||
# Processing configuration
|
||||
batch_size: int = Field(description="Number of records to process in each batch", default=500, ge=100, le=10000)
|
||||
|
||||
quality_threshold: float = Field(
|
||||
description="Minimum quality score required (0.0-1.0)", default=0.8, ge=0.0, le=1.0
|
||||
)
|
||||
|
||||
# Validation settings
|
||||
enable_schema_validation: bool = Field(description="Enable schema validation checks", default=True)
|
||||
|
||||
enable_security_validation: bool = Field(description="Enable security validation checks", default=True)
|
||||
|
||||
enable_quality_validation: bool = Field(description="Enable data quality validation checks", default=True)
|
||||
|
||||
# Transformation options
|
||||
transformations: list[Literal["normalize", "enrich", "aggregate"]] = Field(
|
||||
description="List of transformations to apply", default=["normalize", "enrich"]
|
||||
)
|
||||
|
||||
# Optional description
|
||||
description: str | None = Field(description="Optional description of the processing request", default=None)
|
||||
|
||||
# Test failure scenarios
|
||||
force_validation_failure: bool = Field(
|
||||
description="Force validation failure for testing (demo purposes)", default=False
|
||||
)
|
||||
|
||||
force_transformation_failure: bool = Field(
|
||||
description="Force transformation failure for testing (demo purposes)", default=False
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DataBatch:
|
||||
"""Represents a batch of data being processed."""
|
||||
|
||||
batch_id: str
|
||||
data_type: DataType
|
||||
size: int
|
||||
content: str
|
||||
source: str = "unknown"
|
||||
timestamp: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class ValidationReport:
|
||||
"""Report from data validation."""
|
||||
|
||||
batch_id: str
|
||||
validator_id: str
|
||||
result: ValidationResult
|
||||
issues_found: int
|
||||
processing_time: float
|
||||
details: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class TransformationResult:
|
||||
"""Result from data transformation."""
|
||||
|
||||
batch_id: str
|
||||
transformer_id: str
|
||||
original_size: int
|
||||
processed_size: int
|
||||
transformation_type: str
|
||||
processing_time: float
|
||||
success: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class QualityAssessment:
|
||||
"""Quality assessment result."""
|
||||
|
||||
batch_id: str
|
||||
assessor_id: str
|
||||
quality_score: float
|
||||
recommendations: list[str]
|
||||
processing_time: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProcessingSummary:
|
||||
"""Summary of all processing stages."""
|
||||
|
||||
batch_id: str
|
||||
total_processing_time: float
|
||||
validation_reports: list[ValidationReport]
|
||||
transformation_results: list[TransformationResult]
|
||||
quality_assessments: list[QualityAssessment]
|
||||
final_status: str
|
||||
|
||||
|
||||
# Data Ingestion Stage
|
||||
class DataIngestion(Executor):
|
||||
"""Simulates ingesting data from multiple sources with delays."""
|
||||
|
||||
@handler
|
||||
async def ingest_data(self, request: ProcessingRequest, ctx: WorkflowContext[DataBatch]) -> None:
|
||||
"""Simulate data ingestion with realistic delays based on input configuration."""
|
||||
# Simulate network delay based on data source
|
||||
delay_map = {"database": 1.5, "api": 3.0, "file_upload": 4.0, "streaming": 1.0}
|
||||
delay = delay_map.get(request.data_source, 3.0)
|
||||
await asyncio.sleep(delay) # Fixed delay for demo
|
||||
|
||||
# Simulate data size based on priority and configuration
|
||||
base_size = request.batch_size
|
||||
if request.processing_priority == "critical":
|
||||
size_multiplier = 1.7 # Critical priority gets the largest batches
|
||||
elif request.processing_priority == "high":
|
||||
size_multiplier = 1.3 # High priority gets larger batches
|
||||
elif request.processing_priority == "low":
|
||||
size_multiplier = 0.6 # Low priority gets smaller batches
|
||||
else: # normal
|
||||
size_multiplier = 1.0 # Normal priority uses base size
|
||||
|
||||
actual_size = int(base_size * size_multiplier)
|
||||
|
||||
batch = DataBatch(
|
||||
batch_id=f"batch_{5555}", # Fixed batch ID for demo
|
||||
data_type=DataType(request.data_type),
|
||||
size=actual_size,
|
||||
content=f"Processing {request.data_type} data from {request.data_source}",
|
||||
source=request.data_source,
|
||||
timestamp=asyncio.get_event_loop().time(),
|
||||
)
|
||||
|
||||
# Store both batch data and original request in shared state
|
||||
await ctx.set_shared_state(f"batch_{batch.batch_id}", batch)
|
||||
await ctx.set_shared_state(f"request_{batch.batch_id}", request)
|
||||
|
||||
await ctx.send_message(batch)
|
||||
|
||||
|
||||
# Validation Stage (Fan-out)
|
||||
class SchemaValidator(Executor):
|
||||
"""Validates data schema and structure."""
|
||||
|
||||
@handler
|
||||
async def validate_schema(self, batch: DataBatch, ctx: WorkflowContext[ValidationReport]) -> None:
|
||||
"""Perform schema validation with processing delay."""
|
||||
# Check if schema validation is enabled
|
||||
request = await ctx.get_shared_state(f"request_{batch.batch_id}")
|
||||
if not request or not request.enable_schema_validation:
|
||||
return
|
||||
|
||||
# Simulate schema validation processing
|
||||
processing_time = 2.0 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
# Simulate validation results - consider force failure flag
|
||||
issues = 4 if request.force_validation_failure else 2 # Fixed issue counts
|
||||
|
||||
result = (
|
||||
ValidationResult.VALID
|
||||
if issues <= 1
|
||||
else (ValidationResult.WARNING if issues <= 2 else ValidationResult.ERROR)
|
||||
)
|
||||
|
||||
report = ValidationReport(
|
||||
batch_id=batch.batch_id,
|
||||
validator_id=self.id,
|
||||
result=result,
|
||||
issues_found=issues,
|
||||
processing_time=processing_time,
|
||||
details=f"Schema validation found {issues} issues in {batch.data_type.value} data from {batch.source}",
|
||||
)
|
||||
|
||||
await ctx.send_message(report)
|
||||
|
||||
|
||||
class DataQualityValidator(Executor):
|
||||
"""Validates data quality and completeness."""
|
||||
|
||||
@handler
|
||||
async def validate_quality(self, batch: DataBatch, ctx: WorkflowContext[ValidationReport]) -> None:
|
||||
"""Perform data quality validation."""
|
||||
# Check if quality validation is enabled
|
||||
request = await ctx.get_shared_state(f"request_{batch.batch_id}")
|
||||
if not request or not request.enable_quality_validation:
|
||||
return
|
||||
|
||||
processing_time = 2.5 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
# Quality checks are stricter for higher priority data
|
||||
issues = (
|
||||
2 # Fixed issue count for high priority
|
||||
if request.processing_priority in ["critical", "high"]
|
||||
else 3 # Fixed issue count for normal priority
|
||||
)
|
||||
|
||||
if request.force_validation_failure:
|
||||
issues = max(issues, 4) # Ensure failure
|
||||
|
||||
result = (
|
||||
ValidationResult.VALID
|
||||
if issues <= 1
|
||||
else (ValidationResult.WARNING if issues <= 3 else ValidationResult.ERROR)
|
||||
)
|
||||
|
||||
report = ValidationReport(
|
||||
batch_id=batch.batch_id,
|
||||
validator_id=self.id,
|
||||
result=result,
|
||||
issues_found=issues,
|
||||
processing_time=processing_time,
|
||||
details=f"Quality check found {issues} data quality issues (priority: {request.processing_priority})",
|
||||
)
|
||||
|
||||
await ctx.send_message(report)
|
||||
|
||||
|
||||
class SecurityValidator(Executor):
|
||||
"""Validates data for security and compliance issues."""
|
||||
|
||||
@handler
|
||||
async def validate_security(self, batch: DataBatch, ctx: WorkflowContext[ValidationReport]) -> None:
|
||||
"""Perform security validation."""
|
||||
# Check if security validation is enabled
|
||||
request = await ctx.get_shared_state(f"request_{batch.batch_id}")
|
||||
if not request or not request.enable_security_validation:
|
||||
return
|
||||
|
||||
processing_time = 3.0 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
# Security is more stringent for customer/transaction data
|
||||
issues = 1 if batch.data_type in [DataType.CUSTOMER, DataType.TRANSACTION] else 2
|
||||
|
||||
if request.force_validation_failure:
|
||||
issues = max(issues, 1) # Force at least one security issue
|
||||
|
||||
# Security errors are more serious - less tolerance
|
||||
result = ValidationResult.VALID if issues == 0 else ValidationResult.ERROR
|
||||
|
||||
report = ValidationReport(
|
||||
batch_id=batch.batch_id,
|
||||
validator_id=self.id,
|
||||
result=result,
|
||||
issues_found=issues,
|
||||
processing_time=processing_time,
|
||||
details=f"Security scan found {issues} security issues in {batch.data_type.value} data",
|
||||
)
|
||||
|
||||
await ctx.send_message(report)
|
||||
|
||||
|
||||
# Validation Aggregator (Fan-in)
|
||||
class ValidationAggregator(Executor):
|
||||
"""Aggregates validation results and decides on next steps."""
|
||||
|
||||
@handler
|
||||
async def aggregate_validations(self, reports: list[ValidationReport], ctx: WorkflowContext[DataBatch]) -> None:
|
||||
"""Aggregate all validation reports and make processing decision."""
|
||||
if not reports:
|
||||
return
|
||||
|
||||
batch_id = reports[0].batch_id
|
||||
request = await ctx.get_shared_state(f"request_{batch_id}")
|
||||
|
||||
await asyncio.sleep(1) # Aggregation processing time
|
||||
|
||||
total_issues = sum(report.issues_found for report in reports)
|
||||
has_errors = any(report.result == ValidationResult.ERROR for report in reports)
|
||||
|
||||
# Calculate quality score (0.0 to 1.0)
|
||||
max_possible_issues = len(reports) * 5 # Assume max 5 issues per validator
|
||||
quality_score = max(0.0, 1.0 - (total_issues / max_possible_issues))
|
||||
|
||||
# Decision logic: fail if errors OR quality below threshold
|
||||
should_fail = has_errors or (quality_score < request.quality_threshold)
|
||||
|
||||
if should_fail:
|
||||
failure_reason = []
|
||||
if has_errors:
|
||||
failure_reason.append("validation errors detected")
|
||||
if quality_score < request.quality_threshold:
|
||||
failure_reason.append(
|
||||
f"quality score {quality_score:.2f} below threshold {request.quality_threshold:.2f}"
|
||||
)
|
||||
|
||||
reason = " and ".join(failure_reason)
|
||||
await ctx.add_event(
|
||||
WorkflowCompletedEvent(
|
||||
f"Batch {batch_id} failed validation: {reason}. "
|
||||
f"Total issues: {total_issues}, Quality score: {quality_score:.2f}"
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
# Retrieve original batch from shared state
|
||||
batch_data = await ctx.get_shared_state(f"batch_{batch_id}")
|
||||
if batch_data:
|
||||
await ctx.send_message(batch_data)
|
||||
else:
|
||||
# Fallback: create a simplified batch
|
||||
batch = DataBatch(
|
||||
batch_id=batch_id,
|
||||
data_type=DataType.ANALYTICS,
|
||||
size=500,
|
||||
content="Validated data ready for transformation",
|
||||
)
|
||||
await ctx.send_message(batch)
|
||||
|
||||
|
||||
# Transformation Stage (Fan-out)
|
||||
class DataNormalizer(Executor):
|
||||
"""Normalizes and cleans data."""
|
||||
|
||||
@handler
|
||||
async def normalize_data(self, batch: DataBatch, ctx: WorkflowContext[TransformationResult]) -> None:
|
||||
"""Perform data normalization."""
|
||||
request = await ctx.get_shared_state(f"request_{batch.batch_id}")
|
||||
|
||||
# Check if normalization is enabled
|
||||
if not request or "normalize" not in request.transformations:
|
||||
# Send a "skipped" result
|
||||
result = TransformationResult(
|
||||
batch_id=batch.batch_id,
|
||||
transformer_id=self.id,
|
||||
original_size=batch.size,
|
||||
processed_size=batch.size,
|
||||
transformation_type="normalization",
|
||||
processing_time=0.1,
|
||||
success=True, # Consider skipped as successful
|
||||
)
|
||||
await ctx.send_message(result)
|
||||
return
|
||||
|
||||
processing_time = 4.0 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
# Simulate data size change during normalization
|
||||
processed_size = int(batch.size * 1.0) # No size change for demo
|
||||
|
||||
# Consider force failure flag
|
||||
success = not request.force_transformation_failure # 75% success rate simplified to always success
|
||||
|
||||
result = TransformationResult(
|
||||
batch_id=batch.batch_id,
|
||||
transformer_id=self.id,
|
||||
original_size=batch.size,
|
||||
processed_size=processed_size,
|
||||
transformation_type="normalization",
|
||||
processing_time=processing_time,
|
||||
success=success,
|
||||
)
|
||||
|
||||
await ctx.send_message(result)
|
||||
|
||||
|
||||
class DataEnrichment(Executor):
|
||||
"""Enriches data with additional information."""
|
||||
|
||||
@handler
|
||||
async def enrich_data(self, batch: DataBatch, ctx: WorkflowContext[TransformationResult]) -> None:
|
||||
"""Perform data enrichment."""
|
||||
request = await ctx.get_shared_state(f"request_{batch.batch_id}")
|
||||
|
||||
# Check if enrichment is enabled
|
||||
if not request or "enrich" not in request.transformations:
|
||||
# Send a "skipped" result
|
||||
result = TransformationResult(
|
||||
batch_id=batch.batch_id,
|
||||
transformer_id=self.id,
|
||||
original_size=batch.size,
|
||||
processed_size=batch.size,
|
||||
transformation_type="enrichment",
|
||||
processing_time=0.1,
|
||||
success=True, # Consider skipped as successful
|
||||
)
|
||||
await ctx.send_message(result)
|
||||
return
|
||||
|
||||
processing_time = 5.0 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
processed_size = int(batch.size * 1.3) # Enrichment increases data
|
||||
|
||||
# Consider force failure flag
|
||||
success = not request.force_transformation_failure # 67% success rate simplified to always success
|
||||
|
||||
result = TransformationResult(
|
||||
batch_id=batch.batch_id,
|
||||
transformer_id=self.id,
|
||||
original_size=batch.size,
|
||||
processed_size=processed_size,
|
||||
transformation_type="enrichment",
|
||||
processing_time=processing_time,
|
||||
success=success,
|
||||
)
|
||||
|
||||
await ctx.send_message(result)
|
||||
|
||||
|
||||
class DataAggregator(Executor):
|
||||
"""Aggregates and summarizes data."""
|
||||
|
||||
@handler
|
||||
async def aggregate_data(self, batch: DataBatch, ctx: WorkflowContext[TransformationResult]) -> None:
|
||||
"""Perform data aggregation."""
|
||||
request = await ctx.get_shared_state(f"request_{batch.batch_id}")
|
||||
|
||||
# Check if aggregation is enabled
|
||||
if not request or "aggregate" not in request.transformations:
|
||||
# Send a "skipped" result
|
||||
result = TransformationResult(
|
||||
batch_id=batch.batch_id,
|
||||
transformer_id=self.id,
|
||||
original_size=batch.size,
|
||||
processed_size=batch.size,
|
||||
transformation_type="aggregation",
|
||||
processing_time=0.1,
|
||||
success=True, # Consider skipped as successful
|
||||
)
|
||||
await ctx.send_message(result)
|
||||
return
|
||||
|
||||
processing_time = 2.5 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
processed_size = int(batch.size * 0.5) # Aggregation reduces data
|
||||
|
||||
# Consider force failure flag
|
||||
success = not request.force_transformation_failure # 80% success rate simplified to always success
|
||||
|
||||
result = TransformationResult(
|
||||
batch_id=batch.batch_id,
|
||||
transformer_id=self.id,
|
||||
original_size=batch.size,
|
||||
processed_size=processed_size,
|
||||
transformation_type="aggregation",
|
||||
processing_time=processing_time,
|
||||
success=success,
|
||||
)
|
||||
|
||||
await ctx.send_message(result)
|
||||
|
||||
|
||||
# Quality Assurance Stage (Fan-out)
|
||||
class PerformanceAssessor(Executor):
|
||||
"""Assesses performance characteristics of processed data."""
|
||||
|
||||
@handler
|
||||
async def assess_performance(
|
||||
self, results: list[TransformationResult], ctx: WorkflowContext[QualityAssessment]
|
||||
) -> None:
|
||||
"""Assess performance of transformations."""
|
||||
if not results:
|
||||
return
|
||||
|
||||
batch_id = results[0].batch_id
|
||||
|
||||
processing_time = 2.0 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
avg_processing_time = sum(r.processing_time for r in results) / len(results)
|
||||
success_rate = sum(1 for r in results if r.success) / len(results)
|
||||
|
||||
quality_score = (success_rate * 0.7 + (1 - min(avg_processing_time / 10, 1)) * 0.3) * 100
|
||||
|
||||
recommendations = []
|
||||
if success_rate < 0.8:
|
||||
recommendations.append("Consider improving transformation reliability")
|
||||
if avg_processing_time > 5:
|
||||
recommendations.append("Optimize processing performance")
|
||||
if quality_score < 70:
|
||||
recommendations.append("Review overall data pipeline efficiency")
|
||||
|
||||
assessment = QualityAssessment(
|
||||
batch_id=batch_id,
|
||||
assessor_id=self.id,
|
||||
quality_score=quality_score,
|
||||
recommendations=recommendations,
|
||||
processing_time=processing_time,
|
||||
)
|
||||
|
||||
await ctx.send_message(assessment)
|
||||
|
||||
|
||||
class AccuracyAssessor(Executor):
|
||||
"""Assesses accuracy and correctness of processed data."""
|
||||
|
||||
@handler
|
||||
async def assess_accuracy(
|
||||
self, results: list[TransformationResult], ctx: WorkflowContext[QualityAssessment]
|
||||
) -> None:
|
||||
"""Assess accuracy of transformations."""
|
||||
if not results:
|
||||
return
|
||||
|
||||
batch_id = results[0].batch_id
|
||||
|
||||
processing_time = 3.0 # Fixed processing time
|
||||
await asyncio.sleep(processing_time)
|
||||
|
||||
# Simulate accuracy analysis
|
||||
accuracy_score = 85.0 # Fixed accuracy score
|
||||
|
||||
recommendations = []
|
||||
if accuracy_score < 85:
|
||||
recommendations.append("Review data transformation algorithms")
|
||||
if accuracy_score < 80:
|
||||
recommendations.append("Implement additional validation steps")
|
||||
|
||||
assessment = QualityAssessment(
|
||||
batch_id=batch_id,
|
||||
assessor_id=self.id,
|
||||
quality_score=accuracy_score,
|
||||
recommendations=recommendations,
|
||||
processing_time=processing_time,
|
||||
)
|
||||
|
||||
await ctx.send_message(assessment)
|
||||
|
||||
|
||||
# Final Processing and Completion
|
||||
class FinalProcessor(Executor):
|
||||
"""Final processing stage that combines all results."""
|
||||
|
||||
@handler
|
||||
async def process_final_results(self, assessments: list[QualityAssessment], ctx: WorkflowContext[None]) -> None:
|
||||
"""Generate final processing summary and complete workflow."""
|
||||
if not assessments:
|
||||
await ctx.add_event(WorkflowCompletedEvent("No quality assessments received"))
|
||||
return
|
||||
|
||||
batch_id = assessments[0].batch_id
|
||||
|
||||
# Simulate final processing delay
|
||||
await asyncio.sleep(2)
|
||||
|
||||
# Calculate overall metrics
|
||||
avg_quality_score = sum(a.quality_score for a in assessments) / len(assessments)
|
||||
total_recommendations = sum(len(a.recommendations) for a in assessments)
|
||||
total_processing_time = sum(a.processing_time for a in assessments)
|
||||
|
||||
# Determine final status
|
||||
if avg_quality_score >= 85:
|
||||
final_status = "EXCELLENT"
|
||||
elif avg_quality_score >= 75:
|
||||
final_status = "GOOD"
|
||||
elif avg_quality_score >= 65:
|
||||
final_status = "ACCEPTABLE"
|
||||
else:
|
||||
final_status = "NEEDS_IMPROVEMENT"
|
||||
|
||||
completion_message = (
|
||||
f"Batch {batch_id} processing completed!\n"
|
||||
f"📊 Overall Quality Score: {avg_quality_score:.1f}%\n"
|
||||
f"⏱️ Total Processing Time: {total_processing_time:.1f}s\n"
|
||||
f"💡 Total Recommendations: {total_recommendations}\n"
|
||||
f"🎖️ Final Status: {final_status}"
|
||||
)
|
||||
|
||||
await ctx.add_event(WorkflowCompletedEvent(completion_message))
|
||||
|
||||
|
||||
# Workflow Builder Helper
|
||||
class WorkflowSetupHelper:
|
||||
"""Helper class to set up the complex workflow with shared state management."""
|
||||
|
||||
@staticmethod
|
||||
async def store_batch_data(batch: DataBatch, ctx: WorkflowContext) -> None:
|
||||
"""Store batch data in shared state for later retrieval."""
|
||||
await ctx.set_shared_state(f"batch_{batch.batch_id}", batch)
|
||||
|
||||
|
||||
# Create the workflow instance
|
||||
def create_complex_workflow():
|
||||
"""Create the complex fan-in/fan-out workflow."""
|
||||
# Create all executors
|
||||
data_ingestion = DataIngestion(id="data_ingestion")
|
||||
|
||||
# Validation stage (fan-out)
|
||||
schema_validator = SchemaValidator(id="schema_validator")
|
||||
quality_validator = DataQualityValidator(id="quality_validator")
|
||||
security_validator = SecurityValidator(id="security_validator")
|
||||
validation_aggregator = ValidationAggregator(id="validation_aggregator")
|
||||
|
||||
# Transformation stage (fan-out)
|
||||
data_normalizer = DataNormalizer(id="data_normalizer")
|
||||
data_enrichment = DataEnrichment(id="data_enrichment")
|
||||
data_aggregator_exec = DataAggregator(id="data_aggregator")
|
||||
|
||||
# Quality assurance stage (fan-out)
|
||||
performance_assessor = PerformanceAssessor(id="performance_assessor")
|
||||
accuracy_assessor = AccuracyAssessor(id="accuracy_assessor")
|
||||
|
||||
# Final processing
|
||||
final_processor = FinalProcessor(id="final_processor")
|
||||
|
||||
# Build the workflow with complex fan-in/fan-out patterns
|
||||
return (
|
||||
WorkflowBuilder()
|
||||
.set_start_executor(data_ingestion)
|
||||
# Fan-out to validation stage
|
||||
.add_fan_out_edges(data_ingestion, [schema_validator, quality_validator, security_validator])
|
||||
# Fan-in from validation to aggregator
|
||||
.add_fan_in_edges([schema_validator, quality_validator, security_validator], validation_aggregator)
|
||||
# Fan-out to transformation stage
|
||||
.add_fan_out_edges(validation_aggregator, [data_normalizer, data_enrichment, data_aggregator_exec])
|
||||
# Fan-in to quality assurance stage (both assessors receive all transformation results)
|
||||
.add_fan_in_edges([data_normalizer, data_enrichment, data_aggregator_exec], performance_assessor)
|
||||
.add_fan_in_edges([data_normalizer, data_enrichment, data_aggregator_exec], accuracy_assessor)
|
||||
# Fan-in to final processor
|
||||
.add_fan_in_edges([performance_assessor, accuracy_assessor], final_processor)
|
||||
.build()
|
||||
)
|
||||
|
||||
|
||||
# Export the workflow for DevUI discovery
|
||||
workflow = create_complex_workflow()
|
||||
|
||||
|
||||
def main():
|
||||
"""Launch the fanout workflow in DevUI."""
|
||||
from agent_framework.devui import serve
|
||||
|
||||
# Setup logging
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
logger.info("Starting Complex Fan-In/Fan-Out Data Processing Workflow")
|
||||
logger.info("Available at: http://localhost:8090")
|
||||
logger.info("Entity ID: workflow_complex_workflow")
|
||||
|
||||
# Launch server with the workflow
|
||||
serve(entities=[workflow], port=8090, auto_open=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,71 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Example of using Agent Framework DevUI with in-memory agent registration.
|
||||
|
||||
This demonstrates the simplest way to serve agents as OpenAI-compatible API endpoints.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework import ChatAgent
|
||||
from agent_framework.devui import serve
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
|
||||
def get_weather(
|
||||
location: Annotated[str, "The location to get the weather for."],
|
||||
) -> str:
|
||||
"""Get the weather for a given location."""
|
||||
conditions = ["sunny", "cloudy", "rainy", "stormy"]
|
||||
temperature = 53
|
||||
return f"The weather in {location} is {conditions[0]} with a high of {temperature}°C."
|
||||
|
||||
|
||||
def get_time(
|
||||
timezone: Annotated[str, "The timezone to get time for."] = "UTC",
|
||||
) -> str:
|
||||
"""Get current time for a timezone."""
|
||||
from datetime import datetime
|
||||
|
||||
# Simplified for example
|
||||
return f"Current time in {timezone}: {datetime.now().strftime('%H:%M:%S')}"
|
||||
|
||||
|
||||
def main():
|
||||
"""Main function demonstrating in-memory agent registration."""
|
||||
# Setup logging
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Create agents in code
|
||||
weather_agent = ChatAgent(
|
||||
name="weather-assistant",
|
||||
description="Provides weather information and time",
|
||||
instructions=(
|
||||
"You are a helpful weather and time assistant. Use the available tools to "
|
||||
"provide accurate weather information and current time for any location."
|
||||
),
|
||||
chat_client=OpenAIChatClient(ai_model_id="gpt-4o-mini"),
|
||||
tools=[get_weather, get_time],
|
||||
)
|
||||
|
||||
simple_agent = ChatAgent(
|
||||
name="general-assistant",
|
||||
description="A simple conversational agent",
|
||||
instructions="You are a helpful assistant.",
|
||||
chat_client=OpenAIChatClient(ai_model_id="gpt-4o-mini"),
|
||||
)
|
||||
|
||||
# Collect entities for serving
|
||||
entities = [weather_agent, simple_agent]
|
||||
|
||||
logger.info("Starting DevUI on http://localhost:8090")
|
||||
logger.info("Entity IDs: agent_weather-assistant, agent_general-assistant")
|
||||
|
||||
# Launch server with auto-generated entity IDs
|
||||
serve(entities=entities, port=8090, auto_open=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,7 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Spam detection workflow sample for DevUI testing."""
|
||||
|
||||
from .workflow import workflow
|
||||
|
||||
__all__ = ["workflow"]
|
||||
@@ -0,0 +1,333 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Spam Detection Workflow Sample for DevUI.
|
||||
|
||||
The following sample demonstrates a comprehensive 5-step workflow with multiple executors
|
||||
that process, analyze, detect spam, and handle email messages. This workflow illustrates
|
||||
complex branching logic and realistic processing delays to demonstrate the workflow framework.
|
||||
|
||||
Workflow Steps:
|
||||
1. Email Preprocessor - Cleans and prepares the email
|
||||
2. Content Analyzer - Analyzes email content and structure
|
||||
3. Spam Detector - Determines if the message is spam
|
||||
4a. Spam Handler - Processes spam messages (quarantine, log, remove)
|
||||
4b. Message Responder - Handles legitimate messages (validate, respond)
|
||||
5. Final Processor - Completes the workflow with logging and cleanup
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
|
||||
from agent_framework import (
|
||||
Case,
|
||||
Default,
|
||||
Executor,
|
||||
WorkflowBuilder,
|
||||
WorkflowCompletedEvent,
|
||||
WorkflowContext,
|
||||
handler,
|
||||
)
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
@dataclass
|
||||
class EmailContent:
|
||||
"""A data class to hold the processed email content."""
|
||||
|
||||
original_message: str
|
||||
cleaned_message: str
|
||||
word_count: int
|
||||
has_suspicious_patterns: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class ContentAnalysis:
|
||||
"""A data class to hold content analysis results."""
|
||||
|
||||
email_content: EmailContent
|
||||
sentiment_score: float
|
||||
contains_links: bool
|
||||
has_attachments: bool
|
||||
risk_indicators: list[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class SpamDetectorResponse:
|
||||
"""A data class to hold the spam detection results."""
|
||||
|
||||
analysis: ContentAnalysis
|
||||
is_spam: bool = False
|
||||
confidence_score: float = 0.0
|
||||
spam_reasons: list[str] | None = None
|
||||
|
||||
def __post_init__(self):
|
||||
"""Initialize spam_reasons list if None."""
|
||||
if self.spam_reasons is None:
|
||||
self.spam_reasons = []
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProcessingResult:
|
||||
"""A data class to hold the final processing result."""
|
||||
|
||||
original_message: str
|
||||
action_taken: str
|
||||
processing_time: float
|
||||
status: str
|
||||
is_spam: bool
|
||||
confidence_score: float
|
||||
spam_reasons: list[str]
|
||||
|
||||
|
||||
class EmailRequest(BaseModel):
|
||||
"""Request model for email processing."""
|
||||
|
||||
email: str = Field(
|
||||
description="The email message to be processed.",
|
||||
default="Hi there, are you interested in our new urgent offer today? Click here!",
|
||||
)
|
||||
|
||||
|
||||
class EmailPreprocessor(Executor):
|
||||
"""Step 1: An executor that preprocesses and cleans email content."""
|
||||
|
||||
@handler
|
||||
async def handle_email(self, email: EmailRequest, ctx: WorkflowContext[EmailContent]) -> None:
|
||||
"""Clean and preprocess the email message."""
|
||||
await asyncio.sleep(1.5) # Simulate preprocessing time
|
||||
|
||||
# Simulate email cleaning
|
||||
cleaned = email.email.strip().lower()
|
||||
word_count = len(email.email.split())
|
||||
|
||||
# Check for suspicious patterns
|
||||
suspicious_patterns = ["urgent", "limited time", "act now", "free money"]
|
||||
has_suspicious = any(pattern in cleaned for pattern in suspicious_patterns)
|
||||
|
||||
result = EmailContent(
|
||||
original_message=email.email,
|
||||
cleaned_message=cleaned,
|
||||
word_count=word_count,
|
||||
has_suspicious_patterns=has_suspicious,
|
||||
)
|
||||
|
||||
await ctx.send_message(result)
|
||||
|
||||
|
||||
class ContentAnalyzer(Executor):
|
||||
"""Step 2: An executor that analyzes email content and structure."""
|
||||
|
||||
@handler
|
||||
async def handle_email_content(self, email_content: EmailContent, ctx: WorkflowContext[ContentAnalysis]) -> None:
|
||||
"""Analyze the email content for various indicators."""
|
||||
await asyncio.sleep(2.0) # Simulate analysis time
|
||||
|
||||
# Simulate content analysis
|
||||
sentiment_score = 0.5 if email_content.has_suspicious_patterns else 0.8
|
||||
contains_links = "http" in email_content.cleaned_message or "www" in email_content.cleaned_message
|
||||
has_attachments = "attachment" in email_content.cleaned_message
|
||||
|
||||
# Build risk indicators
|
||||
risk_indicators = []
|
||||
if email_content.has_suspicious_patterns:
|
||||
risk_indicators.append("suspicious_language")
|
||||
if contains_links:
|
||||
risk_indicators.append("contains_links")
|
||||
if has_attachments:
|
||||
risk_indicators.append("has_attachments")
|
||||
if email_content.word_count < 10:
|
||||
risk_indicators.append("too_short")
|
||||
|
||||
analysis = ContentAnalysis(
|
||||
email_content=email_content,
|
||||
sentiment_score=sentiment_score,
|
||||
contains_links=contains_links,
|
||||
has_attachments=has_attachments,
|
||||
risk_indicators=risk_indicators,
|
||||
)
|
||||
|
||||
await ctx.send_message(analysis)
|
||||
|
||||
|
||||
class SpamDetector(Executor):
|
||||
"""Step 3: An executor that determines if a message is spam based on analysis."""
|
||||
|
||||
def __init__(self, spam_keywords: list[str], id: str):
|
||||
"""Initialize the executor with spam keywords."""
|
||||
super().__init__(id=id)
|
||||
self._spam_keywords = spam_keywords
|
||||
|
||||
@handler
|
||||
async def handle_analysis(self, analysis: ContentAnalysis, ctx: WorkflowContext[SpamDetectorResponse]) -> None:
|
||||
"""Determine if the message is spam based on content analysis."""
|
||||
await asyncio.sleep(1.8) # Simulate detection time
|
||||
|
||||
# Check for spam keywords
|
||||
email_text = analysis.email_content.cleaned_message
|
||||
keyword_matches = [kw for kw in self._spam_keywords if kw in email_text]
|
||||
|
||||
# Calculate spam probability
|
||||
spam_score = 0.0
|
||||
spam_reasons = []
|
||||
|
||||
if keyword_matches:
|
||||
spam_score += 0.4
|
||||
spam_reasons.append(f"spam_keywords: {keyword_matches}")
|
||||
|
||||
if analysis.email_content.has_suspicious_patterns:
|
||||
spam_score += 0.3
|
||||
spam_reasons.append("suspicious_patterns")
|
||||
|
||||
if len(analysis.risk_indicators) >= 3:
|
||||
spam_score += 0.2
|
||||
spam_reasons.append("high_risk_indicators")
|
||||
|
||||
if analysis.sentiment_score < 0.4:
|
||||
spam_score += 0.1
|
||||
spam_reasons.append("negative_sentiment")
|
||||
|
||||
is_spam = spam_score >= 0.5
|
||||
|
||||
result = SpamDetectorResponse(
|
||||
analysis=analysis, is_spam=is_spam, confidence_score=spam_score, spam_reasons=spam_reasons
|
||||
)
|
||||
|
||||
await ctx.send_message(result)
|
||||
|
||||
|
||||
class SpamHandler(Executor):
|
||||
"""Step 4a: An executor that handles spam messages with quarantine and logging."""
|
||||
|
||||
@handler
|
||||
async def handle_spam_detection(
|
||||
self,
|
||||
spam_result: SpamDetectorResponse,
|
||||
ctx: WorkflowContext[ProcessingResult],
|
||||
) -> None:
|
||||
"""Handle spam messages by quarantining and logging."""
|
||||
if not spam_result.is_spam:
|
||||
raise RuntimeError("Message is not spam, cannot process with spam handler.")
|
||||
|
||||
await asyncio.sleep(2.2) # Simulate spam handling time
|
||||
|
||||
result = ProcessingResult(
|
||||
original_message=spam_result.analysis.email_content.original_message,
|
||||
action_taken="quarantined_and_logged",
|
||||
processing_time=2.2,
|
||||
status="spam_handled",
|
||||
is_spam=spam_result.is_spam,
|
||||
confidence_score=spam_result.confidence_score,
|
||||
spam_reasons=spam_result.spam_reasons or [],
|
||||
)
|
||||
|
||||
await ctx.send_message(result)
|
||||
|
||||
|
||||
class MessageResponder(Executor):
|
||||
"""Step 4b: An executor that responds to legitimate messages."""
|
||||
|
||||
@handler
|
||||
async def handle_spam_detection(
|
||||
self,
|
||||
spam_result: SpamDetectorResponse,
|
||||
ctx: WorkflowContext[ProcessingResult],
|
||||
) -> None:
|
||||
"""Respond to legitimate messages."""
|
||||
if spam_result.is_spam:
|
||||
raise RuntimeError("Message is spam, cannot respond with message responder.")
|
||||
|
||||
await asyncio.sleep(2.5) # Simulate response time
|
||||
|
||||
result = ProcessingResult(
|
||||
original_message=spam_result.analysis.email_content.original_message,
|
||||
action_taken="responded_and_filed",
|
||||
processing_time=2.5,
|
||||
status="message_processed",
|
||||
is_spam=spam_result.is_spam,
|
||||
confidence_score=spam_result.confidence_score,
|
||||
spam_reasons=spam_result.spam_reasons or [],
|
||||
)
|
||||
|
||||
await ctx.send_message(result)
|
||||
|
||||
|
||||
class FinalProcessor(Executor):
|
||||
"""Step 5: An executor that completes the workflow with final logging and cleanup."""
|
||||
|
||||
@handler
|
||||
async def handle_processing_result(
|
||||
self,
|
||||
result: ProcessingResult,
|
||||
ctx: WorkflowContext[None],
|
||||
) -> None:
|
||||
"""Complete the workflow with final processing and logging."""
|
||||
await asyncio.sleep(1.5) # Simulate final processing time
|
||||
|
||||
total_time = result.processing_time + 1.5
|
||||
|
||||
# Include classification details in completion message
|
||||
classification = "SPAM" if result.is_spam else "LEGITIMATE"
|
||||
reasons = ", ".join(result.spam_reasons) if result.spam_reasons else "none"
|
||||
|
||||
completion_message = (
|
||||
f"Email classified as {classification} (confidence: {result.confidence_score:.2f}). "
|
||||
f"Reasons: {reasons}. "
|
||||
f"Action: {result.action_taken}, "
|
||||
f"Status: {result.status}, "
|
||||
f"Total time: {total_time:.1f}s"
|
||||
)
|
||||
|
||||
await ctx.add_event(WorkflowCompletedEvent(completion_message))
|
||||
|
||||
|
||||
# Create the workflow instance that DevUI can discover
|
||||
spam_keywords = ["spam", "advertisement", "offer", "click here", "winner", "congratulations", "urgent"]
|
||||
|
||||
# Create all the executors for the 5-step workflow
|
||||
email_preprocessor = EmailPreprocessor(id="email_preprocessor")
|
||||
content_analyzer = ContentAnalyzer(id="content_analyzer")
|
||||
spam_detector = SpamDetector(spam_keywords, id="spam_detector")
|
||||
spam_handler = SpamHandler(id="spam_handler")
|
||||
message_responder = MessageResponder(id="message_responder")
|
||||
final_processor = FinalProcessor(id="final_processor")
|
||||
|
||||
# Build the comprehensive 5-step workflow with branching logic
|
||||
workflow = (
|
||||
WorkflowBuilder()
|
||||
.set_start_executor(email_preprocessor)
|
||||
.add_edge(email_preprocessor, content_analyzer)
|
||||
.add_edge(content_analyzer, spam_detector)
|
||||
.add_switch_case_edge_group(
|
||||
spam_detector,
|
||||
[
|
||||
Case(condition=lambda x: x.is_spam, target=spam_handler),
|
||||
Default(target=message_responder),
|
||||
],
|
||||
)
|
||||
.add_edge(spam_handler, final_processor)
|
||||
.add_edge(message_responder, final_processor)
|
||||
.build()
|
||||
)
|
||||
|
||||
# Note: Workflow metadata is determined by executors and graph structure
|
||||
|
||||
|
||||
def main():
|
||||
"""Launch the spam detection workflow in DevUI."""
|
||||
from agent_framework.devui import serve
|
||||
|
||||
# Setup logging
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
logger.info("Starting Spam Detection Workflow")
|
||||
logger.info("Available at: http://localhost:8090")
|
||||
logger.info("Entity ID: workflow_spam_detection")
|
||||
|
||||
# Launch server with the workflow
|
||||
serve(entities=[workflow], port=8090, auto_open=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,7 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Weather agent sample for DevUI testing."""
|
||||
|
||||
from .agent import agent
|
||||
|
||||
__all__ = ["agent"]
|
||||
@@ -0,0 +1,69 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""Sample weather agent for Agent Framework Debug UI."""
|
||||
|
||||
import os
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework import ChatAgent
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
|
||||
def get_weather(
|
||||
location: Annotated[str, "The location to get the weather for."],
|
||||
) -> str:
|
||||
"""Get the weather for a given location."""
|
||||
conditions = ["sunny", "cloudy", "rainy", "stormy"]
|
||||
temperature = 53
|
||||
return f"The weather in {location} is {conditions[0]} with a high of {temperature}°C."
|
||||
|
||||
|
||||
def get_forecast(
|
||||
location: Annotated[str, "The location to get the forecast for."],
|
||||
days: Annotated[int, "Number of days for forecast"] = 3,
|
||||
) -> str:
|
||||
"""Get weather forecast for multiple days."""
|
||||
conditions = ["sunny", "cloudy", "rainy", "stormy"]
|
||||
forecast = []
|
||||
|
||||
for day in range(1, days + 1):
|
||||
condition = conditions[0]
|
||||
temp = 53
|
||||
forecast.append(f"Day {day}: {condition}, {temp}°C")
|
||||
|
||||
return f"Weather forecast for {location}:\n" + "\n".join(forecast)
|
||||
|
||||
|
||||
# Agent instance following Agent Framework conventions
|
||||
agent = ChatAgent(
|
||||
name="WeatherAgent",
|
||||
description="A helpful agent that provides weather information and forecasts",
|
||||
instructions="""
|
||||
You are a weather assistant. You can provide current weather information
|
||||
and forecasts for any location. Always be helpful and provide detailed
|
||||
weather information when asked.
|
||||
""",
|
||||
chat_client=OpenAIChatClient(ai_model_id=os.environ.get("OPENAI_CHAT_MODEL_ID", "gpt-4o")),
|
||||
tools=[get_weather, get_forecast],
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
"""Launch the weather agent in DevUI."""
|
||||
import logging
|
||||
|
||||
from agent_framework.devui import serve
|
||||
|
||||
# Setup logging
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
logger.info("Starting Weather Agent")
|
||||
logger.info("Available at: http://localhost:8090")
|
||||
logger.info("Entity ID: agent_WeatherAgent")
|
||||
|
||||
# Launch server with the agent
|
||||
serve(entities=[agent], port=8090, auto_open=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,7 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Weather agent sample for DevUI testing."""
|
||||
|
||||
from .agent import agent
|
||||
|
||||
__all__ = ["agent"]
|
||||
@@ -0,0 +1,71 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""Sample weather agent for Agent Framework Debug UI."""
|
||||
|
||||
import os
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework import ChatAgent
|
||||
from agent_framework.azure import AzureChatClient
|
||||
|
||||
|
||||
def get_weather(
|
||||
location: Annotated[str, "The location to get the weather for."],
|
||||
) -> str:
|
||||
"""Get the weather for a given location."""
|
||||
conditions = ["sunny", "cloudy", "rainy", "stormy"]
|
||||
temperature = 53
|
||||
return f"The weather in {location} is {conditions[0]} with a high of {temperature}°C."
|
||||
|
||||
|
||||
def get_forecast(
|
||||
location: Annotated[str, "The location to get the forecast for."],
|
||||
days: Annotated[int, "Number of days for forecast"] = 3,
|
||||
) -> str:
|
||||
"""Get weather forecast for multiple days."""
|
||||
conditions = ["sunny", "cloudy", "rainy", "stormy"]
|
||||
forecast = []
|
||||
|
||||
for day in range(1, days + 1):
|
||||
condition = conditions[0]
|
||||
temp = 53
|
||||
forecast.append(f"Day {day}: {condition}, {temp}°C")
|
||||
|
||||
return f"Weather forecast for {location}:\n" + "\n".join(forecast)
|
||||
|
||||
|
||||
# Agent instance following Agent Framework conventions
|
||||
agent = ChatAgent(
|
||||
name="AzureWeatherAgent",
|
||||
description="A helpful agent that provides weather information and forecasts",
|
||||
instructions="""
|
||||
You are a weather assistant. You can provide current weather information
|
||||
and forecasts for any location. Always be helpful and provide detailed
|
||||
weather information when asked.
|
||||
""",
|
||||
chat_client=AzureChatClient(
|
||||
api_key=os.environ.get("AZURE_OPENAI_API_KEY", ""),
|
||||
),
|
||||
tools=[get_weather, get_forecast],
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
"""Launch the Azure weather agent in DevUI."""
|
||||
import logging
|
||||
|
||||
from agent_framework.devui import serve
|
||||
|
||||
# Setup logging
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
logger.info("Starting Azure Weather Agent")
|
||||
logger.info("Available at: http://localhost:8090")
|
||||
logger.info("Entity ID: agent_AzureWeatherAgent")
|
||||
|
||||
# Launch server with the agent
|
||||
serve(entities=[agent], port=8090, auto_open=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user