Introduction:
In the previous parts of this blog series, we explored the Microsoft Agent Framework, AI agents, and how workflows can help automate tasks efficiently. In this Part IV, we dive deeper into creating dynamic workflows with branching logic. By introducing branching, your workflows can make intelligent decisions at runtime, allowing your AI agents to handle more complex scenarios with ease.
Understanding Branching Logic:
Branching logic is a key feature that enables your workflow to choose different paths based on certain conditions. Instead of executing a linear sequence of actions, the workflow evaluates data or message properties and decides which path to follow. This adds flexibility and intelligence, making your AI agents more adaptable to real-world scenarios.
What is Content-Based Routing?
Content-based routing (CBR) is a technique where messages are routed to different destinations based on their content or attributes. The workflow inspects the message and dynamically decides which path or service should handle it. This ensures that each message is processed by the right agent or system according to its specific requirements.
For example:
- A message containing “payment issue” is routed to the Finance agent.
- A message containing “delivery delay” goes to the Logistics agent.
- A message requesting “product information” is sent to the Sales agent.
Content-based routing is a core concept behind conditional edges in Microsoft Agent Framework workflows, allowing dynamic, intelligent decision-making.
Conditional Edges:
Conditional edges are the implementation mechanism for branching logic in workflows. They allow workflows to route messages dynamically based on conditions derived from message content or properties. Combined with content-based routing, they enable:
- Automatic routing of tasks to the appropriate AI agent.
- Adaptation to multiple scenarios without manual intervention.
- Enhanced efficiency and responsiveness across workflows.
Why This Matters for Enterprises:
Dynamic branching and content-based routing have real-world impact for enterprise applications. Workflows with intelligent routing can be applied to:
Customer support systems that automatically direct queries to the right department.
Supply chain management workflows that trigger different actions based on inventory or order conditions.
Internal IT ticketing systems that route issues to specialized teams.
Recruitment processes where applications are dynamically filtered based on skills, experience, or job fit.
By combining AI agents with branching workflows, enterprises can build smart, autonomous systems that reduce manual intervention and improve operational efficiency.
Core Components
- UserSearchDispatcher: Handles initial user input and request processing
- SearchAnalysisExecutor: Analyzes the input and determines search strategy
- ProfessionalSearchExecutor: Performs LinkedIn-focused searches
- SocialSearchExecutor: Searches social media platforms
- UserNotFoundExecutor: Handles cases where users aren't found
Conditional Logic
The workflow uses intelligent decision-making based on:
User Selection Priority: If user specifies a search type, that takes precedence
Content Analysis: Analyzes name length and keywords for auto-detection
Validation Rules: Prevents searches for names that are too short
How Conditional Edges Work
Understanding conditional edges is crucial for building intelligent workflows. In our implementation, we use both conditional edges and multi-selection edge groups to create dynamic routing based on message content and analysis results.
1. Condition Functions
Conditional edges use small logic functions—often called get_condition() or selection functions—that check the content of a message and decide whether a specific path should be followed. These functions act like "traffic signals," returning True or False to control which branch the workflow takes next.
In our workflow, we use a selection function that analyzes the search analysis result:
def select_search_targets(analysis: SearchAnalysisResult, *args, **kwargs):
targets = []
if analysis.search_decision == "Professional":
targets.append("professional_search")
elif analysis.search_decision == "Social":
targets.append("social_search")
elif analysis.search_decision == "Comprehensive":
targets.append("professional_search")
targets.append("social_search")
elif analysis.search_decision == "Not_Found":
targets.append("user_not_found")
return targets
2. Message Inspection
These conditions can look into any part of the message—text, metadata, or even structured data returned by AI agents. In our case, we inspect the SearchAnalysisResult object to examine:
Search Decision: The determined search strategy
User Name: The input name for context
Name Length: For validation purposes
Reason: The logic behind the decision
When agents return data using structured classes (our simple Python classes instead of Pydantic models), the condition function can easily parse and inspect specific fields to make decisions.
3. Defensive Programming
To ensure reliability, each condition function includes error handling. This means that even if the message structure changes or data parsing fails, the workflow won't break—it simply takes a safe fallback route instead.
In our implementation, we handle edge cases:
# Validation in SearchAnalysisExecutor
if name_length < SHORT_NAME_THRESHOLD:
search_decision = "Not_Found"
reason = "Name too short for reliable search"
# Fallback logic
else:
search_decision = "Comprehensive"
reason = "Default comprehensive search"
4. Dynamic Routing
Once the conditions are evaluated, messages are automatically directed to the right destination. In our workflow:
Professional Search: Routes to LinkedIn-focused search executor
Social Search: Routes to social media search executor
Comprehensive Search: Routes to both professional and social executors
Not Found: Routes to error handling executor with suggestions
This enables fully automated, intelligent routing without human intervention, allowing the system to adapt to different types of user input and search requirements.
5. Multi-Selection Edge Groups vs. Conditional Edges
Our implementation uses multi-selection edge groups rather than simple conditional edges because:
Multiple Targets: We can route to multiple executors simultaneously (e.g., both professional and social searches)
Complex Logic: The selection function can return multiple target executors
Flexibility: Easier to add new search types without modifying the workflow structure
.add_multi_selection_edge_group(
search_analysis,
[professional_search, social_search, user_not_found],
selection_func=select_search_targets,
)
This approach provides more flexibility than traditional conditional edges, which typically route to a single destination based on a boolean condition.
Implementation
import asyncio
import os
from dotenv import load_dotenv
from agent_framework import Executor, WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, handler
from typing_extensions import Never
from tavily import TavilyClient
# Load environment variables
load_dotenv()
# Simple data classes without Pydantic
class UserSearchRequest:
def __init__(self, user_name: str, search_type: str = None):
self.user_name = user_name
self.search_type = search_type
class SearchAnalysisResult:
def __init__(self, user_name: str, search_decision: str, reason: str, name_length: int, user_id: str):
self.user_name = user_name
self.search_decision = search_decision
self.reason = reason
self.name_length = name_length
self.user_id = user_id
# Constants
SHORT_NAME_THRESHOLD = 5
LONG_NAME_THRESHOLD = 20
PROFESSIONAL_KEYWORDS = ["ceo", "president", "director", "manager", "engineer", "developer", "architect"]
SOCIAL_KEYWORDS = ["@", "instagram", "twitter", "facebook", "social"]
class UserSearchDispatcher(Executor):
@handler
async def handle(self, user_request: UserSearchRequest, ctx: WorkflowContext[UserSearchRequest]):
if not user_request.user_name or not user_request.user_name.strip():
raise RuntimeError("User name must be a valid non-empty string.")
print(f"🚀 Dispatcher: Starting search for user: {user_request.user_name} (Type: {user_request.search_type or 'Auto-detect'})")
user_id = f"user_{hash(user_request.user_name) % 10000}"
processed_request = UserSearchRequest(
user_name=user_request.user_name.strip(),
search_type=user_request.search_type
)
await ctx.send_message(processed_request)
class SearchAnalysisExecutor(Executor):
@handler
async def handle(self, user_request: UserSearchRequest, ctx: WorkflowContext[SearchAnalysisResult]):
print(f"🔍 Search Analysis: Analyzing search strategy for: {user_request.user_name}")
user_id = f"user_{hash(user_request.user_name) % 10000}"
name_length = len(user_request.user_name)
# Determine search strategy
if name_length < SHORT_NAME_THRESHOLD:
search_decision = "Not_Found"
reason = "Name too short for reliable search"
elif user_request.search_type == "professional":
search_decision = "Professional"
reason = "User selected professional search"
elif user_request.search_type == "social":
search_decision = "Social"
reason = "User selected social search"
elif user_request.search_type == "comprehensive":
search_decision = "Comprehensive"
reason = "User selected comprehensive search"
else:
# Auto-detect based on content
if name_length > LONG_NAME_THRESHOLD:
search_decision = "Comprehensive"
reason = "Long name detected, using comprehensive search"
elif any(keyword in user_request.user_name.lower() for keyword in PROFESSIONAL_KEYWORDS):
search_decision = "Professional"
reason = "Professional title detected"
elif any(keyword in user_request.user_name.lower() for keyword in SOCIAL_KEYWORDS):
search_decision = "Social"
reason = "Social media indicators detected"
else:
search_decision = "Comprehensive"
reason = "Default comprehensive search"
analysis_result = SearchAnalysisResult(
user_name=user_request.user_name,
search_decision=search_decision,
reason=reason,
name_length=name_length,
user_id=user_id
)
await ctx.send_message(analysis_result)
class ProfessionalSearchExecutor(Executor):
@handler
async def handle(self, analysis: SearchAnalysisResult, ctx: WorkflowContext[dict]):
print(f"💼 Professional Search: Searching LinkedIn for: {analysis.user_name}")
try:
api_key = os.getenv("TAVILY_API_KEY")
if not api_key:
raise RuntimeError("TAVILY_API_KEY not found")
client = TavilyClient(api_key=api_key)
search_query = f"{analysis.user_name} LinkedIn profile professional"
response = client.search(
query=search_query,
search_depth="basic",
max_results=3
)
linkedin_results = []
if response.get('results'):
for result in response['results']:
if 'linkedin.com' in result.get('url', '').lower():
linkedin_results.append({
'title': result.get('title', 'N/A'),
'url': result.get('url', 'N/A'),
'content': result.get('content', 'N/A')[:200],
'relevance_score': result.get('score', 0)
})
result = {
"search_type": "professional",
"user_name": analysis.user_name,
"results": linkedin_results,
"summary": f"Found {len(linkedin_results)} professional profiles",
"user_id": analysis.user_id
}
await ctx.yield_output(result)
except Exception as e:
print(f"❌ Professional search error: {str(e)}")
result = {
"search_type": "professional",
"user_name": analysis.user_name,
"results": [],
"summary": f"Error: {str(e)}",
"user_id": analysis.user_id
}
await ctx.yield_output(result)
class SocialSearchExecutor(Executor):
@handler
async def handle(self, analysis: SearchAnalysisResult, ctx: WorkflowContext[dict]):
print(f"📱 Social Search: Searching social media for: {analysis.user_name}")
try:
api_key = os.getenv("TAVILY_API_KEY")
if not api_key:
raise RuntimeError("TAVILY_API_KEY not found")
client = TavilyClient(api_key=api_key)
search_query = f"{analysis.user_name} social media Twitter Instagram Facebook"
response = client.search(
query=search_query,
search_depth="basic",
max_results=3
)
social_results = []
if response.get('results'):
for result in response['results']:
url = result.get('url', '').lower()
if any(platform in url for platform in ['twitter.com', 'instagram.com', 'facebook.com', 'github.com']):
social_results.append({
'platform': 'Twitter' if 'twitter.com' in url else 'Instagram' if 'instagram.com' in url else 'Facebook' if 'facebook.com' in url else 'GitHub',
'title': result.get('title', 'N/A'),
'url': result.get('url', 'N/A'),
'content': result.get('content', 'N/A')[:200],
'relevance_score': result.get('score', 0)
})
result = {
"search_type": "social",
"user_name": analysis.user_name,
"results": social_results,
"summary": f"Found {len(social_results)} social media profiles",
"user_id": analysis.user_id
}
await ctx.yield_output(result)
except Exception as e:
print(f"❌ Social search error: {str(e)}")
result = {
"search_type": "social",
"user_name": analysis.user_name,
"results": [],
"summary": f"Error: {str(e)}",
"user_id": analysis.user_id
}
await ctx.yield_output(result)
class UserNotFoundExecutor(Executor):
@handler
async def handle(self, analysis: SearchAnalysisResult, ctx: WorkflowContext[dict]):
print(f"❌ User Not Found: Handling case for: {analysis.user_name}")
suggestions = [
"Try a different spelling of the name",
"Include middle name or initials",
"Search with professional title",
"Try searching on specific platforms"
]
result = {
"search_type": "not_found",
"user_name": analysis.user_name,
"message": f"User '{analysis.user_name}' not found. Name may be too short or uncommon.",
"suggestions": suggestions,
"user_id": analysis.user_id
}
await ctx.yield_output(result)
class ResultsOutputExecutor(Executor):
@handler
async def handle(self, result: dict, ctx: WorkflowContext[Never, dict]):
print(f"📋 Final Output: Processing results for {result.get('user_name', 'unknown')}")
await ctx.yield_output(result)
# Selection function
def select_search_targets(analysis: SearchAnalysisResult, *args, **kwargs):
targets = []
if analysis.search_decision == "Professional":
targets.append("professional_search")
elif analysis.search_decision == "Social":
targets.append("social_search")
elif analysis.search_decision == "Comprehensive":
targets.append("professional_search") # For simplicity, just do professional
targets.append("social_search")
elif analysis.search_decision == "Not_Found":
targets.append("user_not_found")
return targets
def get_user_input():
print("\n" + "=" * 60)
print("🔍 USER SEARCH INPUT")
print("=" * 60)
# Get user name
while True:
user_name = input("👤 Enter the name to search: ").strip()
if user_name:
break
print("❌ Please enter a valid name.")
# Get search type
print("\n📋 Select search type:")
print("1. Professional Search (LinkedIn focus)")
print("2. Social Search (Social media focus)")
print("3. Comprehensive Search (Full analysis)")
print("4. Auto-detect (Let system decide)")
while True:
try:
choice = input("\n🎯 Enter your choice (1-4): ").strip()
if choice == "1":
search_type = "professional"
break
elif choice == "2":
search_type = "social"
break
elif choice == "3":
search_type = "comprehensive"
break
elif choice == "4":
search_type = None
break
else:
print("❌ Please enter 1, 2, 3, or 4.")
except KeyboardInterrupt:
print("\n👋 Goodbye!")
exit(0)
return UserSearchRequest(user_name=user_name, search_type=search_type)
async def run_interactive_mode(workflow):
print("\n" + "=" * 60)
print("🎮 INTERACTIVE MODE")
print("=" * 60)
user_request = get_user_input()
print(f"\n🚀 Starting search for: {user_request.user_name}")
print(f"🎯 Search type: {user_request.search_type or 'Auto-detect'}")
print("-" * 80)
results_found = False
async for event in workflow.run_stream(user_request):
if isinstance(event, WorkflowOutputEvent):
result = event.data
results_found = True
print(f"\n📊 SEARCH RESULTS:")
print(f"Type: {result.get('search_type', 'unknown')}")
print(f"User: {result.get('user_name', 'unknown')}")
print(f"Summary: {result.get('summary', result.get('message', 'No summary'))}")
if 'results' in result and result['results']:
print(f"\nFound {len(result['results'])} results:")
for i, res in enumerate(result['results'][:3], 1):
print(f"\n{i}. {res.get('title', 'N/A')}")
print(f" URL: {res.get('url', 'N/A')}")
print(f" Score: {res.get('relevance_score', 0):.2f}")
if 'platform' in res:
print(f" Platform: {res['platform']}")
if 'suggestions' in result:
print(f"\n💡 Suggestions:")
for suggestion in result['suggestions']:
print(f" • {suggestion}")
print("\n" + "=" * 80)
if not results_found:
print("\n❌ No results were returned from the workflow.")
print("This might indicate an issue with the workflow execution.")
async def main():
print("=" * 80)
print("🔍 SIMPLE CONDITIONAL USER SEARCH WORKFLOW")
print("=" * 80)
# Create executors
dispatcher = UserSearchDispatcher(id="dispatcher")
search_analysis = SearchAnalysisExecutor(id="search_analysis")
professional_search = ProfessionalSearchExecutor(id="professional_search")
social_search = SocialSearchExecutor(id="social_search")
user_not_found = UserNotFoundExecutor(id="user_not_found")
# Build workflow
workflow = (
WorkflowBuilder()
.set_start_executor(dispatcher)
.add_edge(dispatcher, search_analysis)
.add_multi_selection_edge_group(
search_analysis,
[professional_search, social_search, user_not_found],
selection_func=select_search_targets,
)
.build()
)
# Run interactive mode
await run_interactive_mode(workflow)
print("\n👋 Thank you for using the Simple Conditional User Search Workflow!")
if __name__ == "__main__":
asyncio.run(main())
Output
Conclusion:
Branching logic, conditional edges, and content-based routing empower the Microsoft Agent Framework to handle complex, real-world workflows. By making runtime decisions, your AI agents become intelligent, adaptive, and enterprise-ready. In the next part of this series, we will explore advanced examples of AI agents using branching workflows and how they can integrate with external systems for even greater capabilities.
The Conditional User Search Workflow demonstrates the power of the Microsoft Agent Framework for building sophisticated, conditional workflows. By combining:
- Interactive user input
- Intelligent routing logic
- Real-time web search
- Formatted output display
We created a robust system that can adapt to different search requirements while providing a clean user experience.
The key to success was understanding the framework's capabilities and working within its constraints, particularly avoiding Pydantic compatibility issues while maintaining clean, maintainable code.
This workflow serves as a foundation for more complex search and analysis systems, demonstrating how conditional logic can create intelligent, user-friendly applications.
Thanks
Sreeni Ramadorai
Top comments (0)