Real-World Examples of Multi-Task Orchestration
Real-World Examples of Multi-Task Orchestration
The Fiscus SDK provides powerful orchestration capabilities to streamline complex workflows across multiple APIs. Here, we present three comprehensive examples that demonstrate different multi-task orchestration scenarios using the latest orchestration structure.
- Python
Example 1: Real-Time Content Moderation Workflow
In this example, we’ll create a workflow to moderate user-uploaded content. The workflow will:
- Analyze content sentiment.
- Flag inappropriate content for review if detected.
- Notify an admin if flagged content needs immediate attention.
Step-by-Step Breakdown
- Analyze Content Sentiment – Uses a content analysis API to assess sentiment and detect offensive language.
- Flag Content for Review – If content is deemed offensive, it’s flagged for human review.
- Admin Notification – If flagged content has high severity, an admin is notified immediately.
Code Walkthrough
Define the tasks for content moderation.
tasks = [
{
'id': 'analyze_content',
'connector': 'ContentAPI',
'operation': 'analyze_sentiment',
'params': {
'content': 'context.user_content'
},
'onSuccess': {
'updateContext': {
'sentiment': 'response.result.sentiment',
'offensive': 'response.result.offensive'
}
}
},
{
'id': 'flag_inappropriate_content',
'connector': 'ModerationAPI',
'operation': 'flag_content',
'params': {
'content_id': 'context.content_id'
},
'conditions': [
{
'type': 'if',
'expression': 'context.offensive == true',
'nestedTasks': [
{
'id': 'check_severity',
'connector': 'AnalysisAPI',
'operation': 'check_severity',
'params': {
'content_id': 'context.content_id'
},
'onSuccess': {
'updateContext': {
'severity': 'response.result.severity'
}
}
},
{
'id': 'notify_admin',
'connector': 'NotificationAPI',
'operation': 'send_alert',
'params': {
'admin_id': 'admin_001',
'message': 'Immediate review required for high-severity content.'
},
'conditions': [
{
'type': 'if',
'expression': 'context.severity == "high"'
}
]
}
]
}
]
}
]
response = client.execute(tasks=tasks, user=user)
if response.success:
print("Content moderation workflow completed successfully!")
else:
print(f"Error in content moderation workflow: {response.error_message}")
What’s Happening
- Sentiment Analysis: The content is analyzed for sentiment and offensive language, updating context with relevant results.
- Flagging Content: If offensive language is detected, a nested task checks its severity.
- Admin Alert: If the severity is high, an immediate admin notification is sent, ensuring rapid response.
Example 2: Multi-Step Order Verification Workflow in Finance
This example automates the order verification process in a finance application, where:
- Order details are verified.
- Fraud checks are conducted based on order data.
- The user is notified of the order status.
Step-by-Step Breakdown
- Verify Order Details – Pulls order details to ensure completeness.
- Conduct Fraud Check – Runs a fraud detection scan if the order meets certain criteria.
- Notify User – Sends an email notification regarding the status of the order.
Code Walkthrough
Define the tasks for order verification and fraud checking.
tasks = [
{
'id': 'verify_order_details',
'connector': 'FinanceAPI',
'operation': 'fetch_order_details',
'params': {
'order_id': 'context.order_id'
},
'onSuccess': {
'updateContext': {
'order_verified': 'response.result.verified'
}
}
},
{
'id': 'fraud_check',
'connector': 'FraudDetectionAPI',
'operation': 'run_fraud_check',
'params': {
'order_id': 'context.order_id'
},
'conditions': [
{
'type': 'if',
'expression': 'context.order_verified == true',
'nestedTasks': [
{
'id': 'notify_user_fraud_check',
'connector': 'NotificationAPI',
'operation': 'send_email',
'params': {
'to': 'context.user_email',
'subject': 'Order Verification',
'body': 'Your order is under review for security checks.'
}
}
]
}
],
'onSuccess': {
'updateContext': {
'fraud_check_complete': 'response.result.complete',
'fraudulent': 'response.result.fraudulent'
}
}
},
{
'id': 'final_order_status',
'connector': 'NotificationAPI',
'operation': 'send_email',
'params': {
'to': 'context.user_email',
'subject': 'Order Status',
'body': 'Your order has been verified successfully.' if 'context.fraudulent == false' else 'Please contact support regarding your order.'
},
'conditions': [
{
'type': 'if',
'expression': 'context.fraud_check_complete == true'
}
]
}
]
response = client.execute(tasks=tasks, user=user)
if response.success:
print("Order verification workflow completed successfully!")
else:
print(f"Error in order verification workflow: {response.error_message}")
What’s Happening
- Order Verification: Order details are fetched and checked for completeness.
- Fraud Detection: A fraud check runs if the order is verified, with a follow-up email to the user.
- Final Status Notification: Once the fraud check completes, the user receives an order status notification, which adapts based on whether fraud was detected.
Example 3: Integrating Real-Time Weather Data into Delivery Routing
For logistics, this workflow uses real-time weather data to adjust delivery routing. The process:
- Retrieves current weather along the delivery route.
- Adjusts the delivery route if severe weather is detected.
- Updates the customer with an estimated time of arrival (ETA) based on the revised route.
Step-by-Step Breakdown
- Fetch Weather Conditions – Gets weather conditions along the planned route.
- Adjust Route for Safety – If severe weather is detected, a new route is planned.
- Notify Customer – Sends the customer an updated ETA.
Code Walkthrough
Define the tasks for weather data integration and route adjustment.
tasks = [
{
'id': 'fetch_weather_conditions',
'connector': 'WeatherAPI',
'operation': 'get_route_weather',
'params': {
'route_id': 'context.route_id'
},
'onSuccess': {
'updateContext': {
'severe_weather': 'response.result.severe_weather'
}
}
},
{
'id': 'adjust_route',
'connector': 'RoutingAPI',
'operation': 'recalculate_route',
'params': {
'route_id': 'context.route_id'
},
'conditions': [
{
'type': 'if',
'expression': 'context.severe_weather == true',
'nestedTasks': [
{
'id': 'notify_delivery_team',
'connector': 'NotificationAPI',
'operation': 'send_alert',
'params': {
'team_id': 'delivery_team',
'message': 'Route has been adjusted due to severe weather conditions.'
}
}
]
}
],
'onSuccess': {
'updateContext': {
'route_adjusted': 'response.result.route_adjusted'
}
}
},
{
'id': 'notify_customer_eta',
'connector': 'NotificationAPI',
'operation': 'send_eta',
'params': {
'customer_id': 'context.customer_id',
'message': 'Your delivery ETA has been updated based on current route conditions.'
},
'conditions': [
{
'type': 'if',
'expression': 'context.route_adjusted == true'
}
]
}
]
response = client.execute(tasks=tasks, user=user)
if response.success:
print("Delivery routing workflow completed successfully!")
else:
print(f"Error in delivery routing workflow: {response.error_message}")
What’s Happening
- Weather Conditions: Weather data is gathered along the delivery route, with severe conditions flagged in the context.
- Route Adjustment: If severe weather is detected, the route is recalculated, and the delivery team is alerted.
- Customer Notification: The customer receives an updated ETA message based on the adjusted route, ensuring accurate delivery timing.
Final Thoughts
These examples illustrate the power and flexibility of Fiscus SDK for orchestrating complex workflows across industries. Whether handling real-time data, performing automated checks, or routing information based on conditional logic, Fiscus allows developers to streamline tasks and build robust, adaptive workflows to meet diverse operational needs.