[FEATURE] structured output in Graphs

[ibaki-aws]

Problem Statement

When working with Graphs it's not currently possible to have conditional edge traversal based on the structured output of a node result

Proposed Solution

No response

Use Case

Feedback loops and executor / reviewer scenarios

Alternatives Solutions

No response

Additional Context

No response

[pitta-bread]

Possibly a different feature request, but wanted to raise the idea and seems very linked. Langgraph has a way of doing dynamic fan out, via a Send API. It enables interesting and efficient flows, specifically when the number of parallel edges in the graph is not know ahead of runtime.

Perhaps something similar is required here?
Interested in the maintainers' thoughts.

https://langchain-ai.github.io/langgraph/concepts/low_level/?h=send#send

[JackYPCOnline]

Bump this up to medium high

[afarntrog]

@ibaki-aws please confirm if you're looking for something along the following lines:
Example of routing based on structured output (note - this is not yet supported but will be once we get confirmation). This introduces structured output on the agent and graph level and allows for taking action (such as routing) based on that structured response:

"""
Example: Conditional Edge Traversal Based on Structured Output in Graphs

The implementation allows you to:
1. Define a Pydantic model for structured output
2. Pass it to the Graph when invoking
3. Access the structured output in conditional edge functions via GraphState
4. Route to different nodes based on the structured output content
"""

from pydantic import BaseModel
from typing import Literal

from strands import Agent
from strands.multiagent.graph import GraphBuilder, GraphState


# Step 1: Define a Pydantic model for structured output
class ClassificationResult(BaseModel):
    """Structured output from the classifier agent."""
    category: Literal["technical", "billing", "general"]
    confidence: float
    reasoning: str


# Step 2: Create agents for each node
classifier_agent = Agent(
    name="classifier",
    system_prompt="You are a customer support classifier. Analyze the incoming query and classify it.",
)

technical_agent = Agent(
    name="technical_support",
    system_prompt="You are a technical support specialist. Help with technical issues.",
)

billing_agent = Agent(
    name="billing_support",
    system_prompt="You are a billing specialist. Help with billing and payment issues.",
)

general_agent = Agent(
    name="general_support",
    system_prompt="You are a general support agent. Handle general inquiries.",
)



# Step 3: Define conditional edge functions that access structured output
def route_to_technical(state: GraphState) -> bool:
    """Route to technical agent if classification is 'technical'."""
    classifier_result = state.results.get("classifier")
    if classifier_result and classifier_result.result:
        # Access the structured_output from the AgentResult
        structured_result = classifier_result.result.structured_output
        if structured_result and isinstance(structured_result, ClassificationResult):
            return structured_result.category == "technical"
    return False


def route_to_billing(state: GraphState) -> bool:
    """Route to billing agent if classification is 'billing'."""
    classifier_result = state.results.get("classifier")
    if classifier_result and classifier_result.result:
        # Access the structured_output from the AgentResult
        structured_result = classifier_result.result.structured_output
        if structured_result and isinstance(structured_result, ClassificationResult):
            return structured_result.category == "billing"
    return False


def route_to_general(state: GraphState) -> bool:
    """Route to general agent if classification is 'general'."""
    classifier_result = state.results.get("classifier")
    if classifier_result and classifier_result.result:
        # Access the structured_output from the AgentResult
        structured_result = classifier_result.result.structured_output
        if structured_result and isinstance(structured_result, ClassificationResult):
            return structured_result.category == "general"
    return False


# Step 4: Build the graph with conditional edges
def create_support_routing_graph():
    """Create a customer support routing graph with conditional edges."""
    builder = GraphBuilder()
    
    # Add nodes
    builder.add_node(classifier_agent, "classifier")
    builder.add_node(technical_agent, "technical")
    builder.add_node(billing_agent, "billing")
    builder.add_node(general_agent, "general")
    
    # Add conditional edges from classifier to specialized agents
    builder.add_edge("classifier", "technical", condition=route_to_technical)
    builder.add_edge("classifier", "billing", condition=route_to_billing)
    builder.add_edge("classifier", "general", condition=route_to_general)
    
    # Set entry point
    builder.set_entry_point("classifier")
    
    # Build with execution limits
    builder.set_max_node_executions(5)
    builder.set_execution_timeout(60.0)
    
    return builder.build()


# Step 5: Execute the graph with structured output model
def run_example():
    """Run the example to show conditional routing based on structured output."""
    graph = create_support_routing_graph()
    
    # Example queries
    queries = [
        "My application keeps crashing when I try to upload files",  # Should route to technical
        "I was charged twice for my subscription",                   # Should route to billing  
        "What are your business hours?",                             # Should route to general
    ]
    
    for query in queries:
        print(f"\n{'='*60}")
        print(f"Query: {query}")
        print('='*60)
        
        # Execute with structured_output_model for conditional routing
        result = graph(query, structured_output_model=ClassificationResult)
        
        print(f"Status: {result.status}")
        print(f"Execution order: {[n.node_id for n in result.execution_order]}")
        
        # Show the classification result
        classifier_result = result.results.get("classifier")
        if classifier_result and classifier_result.result:
            structured_result = classifier_result.result.structured_output
            if structured_result:
                print(f"Classification: {structured_result.category} (confidence: {structured_result.confidence})")
                print(f"Reasoning: {structured_result.reasoning}")

run_example()

Output:

============================================================
Query: My application keeps crashing when I try to upload files
============================================================
I'll analyze your query and classify it appropriately.
Tool #1: ClassificationResult
I'll analyze your query about the application crashing during file uploads and classify it appropriately.
Tool #1: ClassificationResult
Status: Status.COMPLETED
Execution order: ['classifier', 'technical']
Classification: technical (confidence: 0.95)
Reasoning: This query describes a specific technical problem - application crashes during file upload functionality. This is clearly a technical issue that would require troubleshooting steps, error analysis, or technical support to resolve. The mention of "crashing" and specific functionality (file upload) indicates a technical malfunction rather than a billing inquiry or general information request.

============================================================
Query: I was charged twice for my subscription
============================================================
I'll analyze your query and classify it appropriately.
Tool #2: ClassificationResult
I understand you were charged twice for your subscription - this is definitely a billing issue that I can help you resolve.

Let me first classify this issue properly and then we can work on getting this resolved for you.
Tool #1: ClassificationResult
Status: Status.COMPLETED
Execution order: ['classifier', 'billing']
Classification: billing (confidence: 0.98)
Reasoning: This query is clearly about a billing issue - the customer reports being charged twice for their subscription. This involves payment processing, account charges, and subscription billing, which are all billing-related concerns. The customer will likely need assistance with refunds, charge verification, or billing account review, all of which fall under billing support.

============================================================
Query: What are your business hours?
============================================================
I'll analyze your query and classify it appropriately.
Tool #3: ClassificationResult
I'll help you with information about our business hours.

Our standard business hours are:
- **Monday - Friday**: 9:00 AM to 6:00 PM
- **Saturday**: 10:00 AM to 4:00 PM  
- **Sunday**: Closed

Please note that these hours may vary during holidays or special occasions. For the most up-to-date information or if you need assistance outside these hours, you can also:
- Use our online support portal available 24/7
- Leave a message and we'll get back to you during business hours
- Check our website for any temporary hour changes

Is there anything specific you need help with regarding our services during these hours?
Tool #1: ClassificationResult
Status: Status.COMPLETED
Execution order: ['classifier', 'general']
Classification: general (confidence: 0.92)
Reasoning: This query is asking for general information about business hours. It doesn't involve any technical issues, problems with products/services, or billing concerns. This is a straightforward informational request about company operations that would typically be handled by general customer service or found in basic company information.

Notice the Execution Order for each of the 3 queries:
Execution order: ['classifier', 'technical']

[ibaki-aws]

@ibaki-aws please confirm if you're looking for something along the following lines: Example of routing based on structured output (note - this is not yet supported but will be once we get confirmation). This introduces structured output on the agent and graph level and allows for taking action (such as routing) based on that structured response:

"""
Example: Conditional Edge Traversal Based on Structured Output in Graphs

The implementation allows you to:

1. Define a Pydantic model for structured output
2. Pass it to the Graph when invoking
3. Access the structured output in conditional edge functions via GraphState
4. Route to different nodes based on the structured output content
   """

from pydantic import BaseModel
from typing import Literal

from strands import Agent
from strands.multiagent.graph import GraphBuilder, GraphState

Step 1: Define a Pydantic model for structured output

class ClassificationResult(BaseModel):
"""Structured output from the classifier agent."""
category: Literal["technical", "billing", "general"]
confidence: float
reasoning: str

Step 2: Create agents for each node

classifier_agent = Agent(
name="classifier",
system_prompt="You are a customer support classifier. Analyze the incoming query and classify it.",
)

technical_agent = Agent(
name="technical_support",
system_prompt="You are a technical support specialist. Help with technical issues.",
)

billing_agent = Agent(
name="billing_support",
system_prompt="You are a billing specialist. Help with billing and payment issues.",
)

general_agent = Agent(
name="general_support",
system_prompt="You are a general support agent. Handle general inquiries.",
)

Step 3: Define conditional edge functions that access structured output

def route_to_technical(state: GraphState) -> bool:
"""Route to technical agent if classification is 'technical'."""
classifier_result = state.results.get("classifier")
if classifier_result and classifier_result.result:
# Access the structured_output from the AgentResult
structured_result = classifier_result.result.structured_output
if structured_result and isinstance(structured_result, ClassificationResult):
return structured_result.category == "technical"
return False

def route_to_billing(state: GraphState) -> bool:
"""Route to billing agent if classification is 'billing'."""
classifier_result = state.results.get("classifier")
if classifier_result and classifier_result.result:
# Access the structured_output from the AgentResult
structured_result = classifier_result.result.structured_output
if structured_result and isinstance(structured_result, ClassificationResult):
return structured_result.category == "billing"
return False

def route_to_general(state: GraphState) -> bool:
"""Route to general agent if classification is 'general'."""
classifier_result = state.results.get("classifier")
if classifier_result and classifier_result.result:
# Access the structured_output from the AgentResult
structured_result = classifier_result.result.structured_output
if structured_result and isinstance(structured_result, ClassificationResult):
return structured_result.category == "general"
return False

Step 4: Build the graph with conditional edges

def create_support_routing_graph():
"""Create a customer support routing graph with conditional edges."""
builder = GraphBuilder()

# Add nodes
builder.add_node(classifier_agent, "classifier")
builder.add_node(technical_agent, "technical")
builder.add_node(billing_agent, "billing")
builder.add_node(general_agent, "general")

# Add conditional edges from classifier to specialized agents
builder.add_edge("classifier", "technical", condition=route_to_technical)
builder.add_edge("classifier", "billing", condition=route_to_billing)
builder.add_edge("classifier", "general", condition=route_to_general)

# Set entry point
builder.set_entry_point("classifier")

# Build with execution limits
builder.set_max_node_executions(5)
builder.set_execution_timeout(60.0)

return builder.build()

Step 5: Execute the graph with structured output model

def run_example():
"""Run the example to show conditional routing based on structured output."""
graph = create_support_routing_graph()

# Example queries
queries = [
    "My application keeps crashing when I try to upload files",  # Should route to technical
    "I was charged twice for my subscription",                   # Should route to billing  
    "What are your business hours?",                             # Should route to general
]

for query in queries:
    print(f"\n{'='*60}")
    print(f"Query: {query}")
    print('='*60)
    
    # Execute with structured_output_model for conditional routing
    result = graph(query, structured_output_model=ClassificationResult)
    
    print(f"Status: {result.status}")
    print(f"Execution order: {[n.node_id for n in result.execution_order]}")
    
    # Show the classification result
    classifier_result = result.results.get("classifier")
    if classifier_result and classifier_result.result:
        structured_result = classifier_result.result.structured_output
        if structured_result:
            print(f"Classification: {structured_result.category} (confidence: {structured_result.confidence})")
            print(f"Reasoning: {structured_result.reasoning}")

run_example()
Output:

============================================================
Query: My application keeps crashing when I try to upload files
============================================================
I'll analyze your query and classify it appropriately.
Tool #1: ClassificationResult
I'll analyze your query about the application crashing during file uploads and classify it appropriately.
Tool #1: ClassificationResult
Status: Status.COMPLETED
Execution order: ['classifier', 'technical']
Classification: technical (confidence: 0.95)
Reasoning: This query describes a specific technical problem - application crashes during file upload functionality. This is clearly a technical issue that would require troubleshooting steps, error analysis, or technical support to resolve. The mention of "crashing" and specific functionality (file upload) indicates a technical malfunction rather than a billing inquiry or general information request.

============================================================
Query: I was charged twice for my subscription
============================================================
I'll analyze your query and classify it appropriately.
Tool #2: ClassificationResult
I understand you were charged twice for your subscription - this is definitely a billing issue that I can help you resolve.

Let me first classify this issue properly and then we can work on getting this resolved for you.
Tool #1: ClassificationResult
Status: Status.COMPLETED
Execution order: ['classifier', 'billing']
Classification: billing (confidence: 0.98)
Reasoning: This query is clearly about a billing issue - the customer reports being charged twice for their subscription. This involves payment processing, account charges, and subscription billing, which are all billing-related concerns. The customer will likely need assistance with refunds, charge verification, or billing account review, all of which fall under billing support.

============================================================
Query: What are your business hours?
============================================================
I'll analyze your query and classify it appropriately.
Tool #3: ClassificationResult
I'll help you with information about our business hours.

Our standard business hours are:
- **Monday - Friday**: 9:00 AM to 6:00 PM
- **Saturday**: 10:00 AM to 4:00 PM  
- **Sunday**: Closed

Please note that these hours may vary during holidays or special occasions. For the most up-to-date information or if you need assistance outside these hours, you can also:
- Use our online support portal available 24/7
- Leave a message and we'll get back to you during business hours
- Check our website for any temporary hour changes

Is there anything specific you need help with regarding our services during these hours?
Tool #1: ClassificationResult
Status: Status.COMPLETED
Execution order: ['classifier', 'general']
Classification: general (confidence: 0.92)
Reasoning: This query is asking for general information about business hours. It doesn't involve any technical issues, problems with products/services, or billing concerns. This is a straightforward informational request about company operations that would typically be handled by general customer service or found in basic company information.

Notice the Execution Order for each of the 3 queries: Execution order: ['classifier', 'technical']

Yes, exactly!

[afarntrog]

This feature is currently supported 🎉 Please see the following example:

"""
Example: Conditional Edge Traversal Based on Structured Output in Graphs
"""

from pydantic import BaseModel
from typing import Literal

from strands import Agent
from strands.multiagent.graph import GraphBuilder, GraphState


# Step 1: Define a Pydantic model for structured output
class ClassificationResult(BaseModel):
    """Structured output from the classifier agent."""
    category: Literal["technical", "billing", "general"]
    confidence: float
    reasoning: str


# Step 2: Create agents for each node
classifier_agent = Agent(
    name="classifier",
    system_prompt="You are a customer support classifier. Analyze the incoming query and classify it.",
    structured_output_model=ClassificationResult # <-- Note the structured output passed to this `classifier_agent`
)

technical_agent = Agent(
    name="technical_support",
    system_prompt="You are a technical support specialist. Help with technical issues.",
)

billing_agent = Agent(
    name="billing_support",
    system_prompt="You are a billing specialist. Help with billing and payment issues.",
)

general_agent = Agent(
    name="general_support",
    system_prompt="You are a general support agent. Handle general inquiries.",
)



# Step 3: Define conditional edge functions that access structured output
def route_to_technical(state: GraphState) -> bool:
    """Route to technical agent if classification is 'technical'."""
    classifier_result = state.results.get("classifier")
    if classifier_result and classifier_result.result:
        # Access the structured_output from the AgentResult
        structured_result = classifier_result.result.structured_output
        if structured_result and isinstance(structured_result, ClassificationResult):
            return structured_result.category == "technical"
    return False


def route_to_billing(state: GraphState) -> bool:
    """Route to billing agent if classification is 'billing'."""
    classifier_result = state.results.get("classifier")
    if classifier_result and classifier_result.result:
        # Access the structured_output from the AgentResult
        structured_result = classifier_result.result.structured_output
        if structured_result and isinstance(structured_result, ClassificationResult):
            return structured_result.category == "billing"
    return False


def route_to_general(state: GraphState) -> bool:
    """Route to general agent if classification is 'general'."""
    classifier_result = state.results.get("classifier")
    if classifier_result and classifier_result.result:
        # Access the structured_output from the AgentResult
        structured_result = classifier_result.result.structured_output
        if structured_result and isinstance(structured_result, ClassificationResult):
            return structured_result.category == "general"
    return False


# Step 4: Build the graph with conditional edges
def create_support_routing_graph():
    """Create a customer support routing graph with conditional edges."""
    builder = GraphBuilder()
    
    # Add nodes
    builder.add_node(classifier_agent, "classifier")
    builder.add_node(technical_agent, "technical")
    builder.add_node(billing_agent, "billing")
    builder.add_node(general_agent, "general")
    
    # Add conditional edges from classifier to specialized agents
    builder.add_edge("classifier", "technical", condition=route_to_technical)
    builder.add_edge("classifier", "billing", condition=route_to_billing)
    builder.add_edge("classifier", "general", condition=route_to_general)
    
    # Set entry point
    builder.set_entry_point("classifier")
    
    # Build with execution limits
    builder.set_max_node_executions(5)
    builder.set_execution_timeout(60.0)
    
    return builder.build()


# Step 5: Execute the graph with structured output model
def run_example():
    """Run the example to show conditional routing based on structured output."""
    graph = create_support_routing_graph()
    
    # Example queries
    queries = [
        "My application keeps crashing when I try to upload files",  # Should route to technical
        # "I was charged twice for my subscription",                   # Should route to billing  
        # "What are your business hours?",                             # Should route to general
    ]
    
    for query in queries:
        print(f"\n{'='*60}")
        print(f"Query: {query}")
        print('='*60)
        
        result = graph(query)
        
        print(f"Status: {result.status}")
        print(f"Execution order: {[n.node_id for n in result.execution_order]}")
        
        # Show the classification result
        classifier_result = result.results.get("classifier")
        if classifier_result and classifier_result.result:
            structured_result = classifier_result.result.structured_output
            if structured_result:
                print(f"Classification: {structured_result.category} (confidence: {structured_result.confidence})")
                print(f"Reasoning: {structured_result.reasoning}")