UserNode

UserNode#

UserNode is a node that receives user input from the terminal and provides interactive functionality for the graph.

Overview#

UserNode inherits from AsyncNode, providing a simple user interaction interface. When the graph execution reaches UserNode, it displays a prompt in the terminal and waits for user input, then passes the input to the subsequent flow.

Characteristics#

  • Asynchronous Processing: Avoid blocking the event loop, support high-concurrency scenarios

  • Configurable Prompt: Supports custom user prompts

  • Flexible Storage: Can specify the storage key name for input in the state

  • Exception Safety: Gracefully handle various exceptions

  • Simple and Easy to Use: Simple configuration, Ready to use out of the box

Basic Usage#

from evofabric.core.agent._user_node import UserNode

# Create a UserNode with default configuration
user_node = UserNode()

# Custom prompt message
custom_node = UserNode(prompt_message="Please enter your command: ")

# Custom storage key name
keyed_node = UserNode(input_key="custom_input")

Use in the figure#

# Define a graph containing UserNode
from evofabric.core.graph import GraphBuilder

graph = GraphBuilder()

# Add nodes and edges
graph.add_node("user_input", User  Node(prompt_message="Please enter task description: "))
graph.add_node("process", YourProcessingNode())
graph.add_edge("user_input", "process")

# Execute the graph
final_state = await graph.run({"messages": []})

# Assume user input "Complete data analysis"
# Result: final_state["messages"] contains the user input message

Best Practices#

1. Input Validation

If user input validation is required, processing logic can be added in subsequent nodes:

async def validate_input(state: State) -> StateDelta:
    user_input = state.get("user_input", "")
    if not user_input.strip():
        return {"error": "Input cannot be empty"}
    # Other validation ...

2. Error Handling

UserNode already has built-in comprehensive exception handling, but it is recommended to add business logic error handling in the subsequent nodes of the graph:

async def handle_error(state: State) -> StateDelta:
    if "error" in state:
        print(f"Error: {state['error']}")
        return {"messages": [UserMessage(content="Please re-enter")]}
    return {}

3. Multi-turn Dialogue

Can implement a simple multi-turn dialogue mode:

class ConversationNode(AsyncNode):
    async def __call__(self, state: State) -> StateDelta:
        messages = state.get("  "messages", [])
        if not messages:
            # First round: ask for requirements
            return {"messages": [UserMessage(content="What help do you need?")]}
        else:
            # Subsequent rounds: process user replies
            last_message = messages[-1].content
            if last_message.lower() in ["exit", "quit"]:
                return {"messages": [UserMessage(content="Goodbye!")]}
            # Process other replies...
            return {"messages": [UserMessage(content="Your reply has been received")]}

Parameter Description#

evofabric.core.agent.prompt_message#

Displayed user input prompt

Type:

str

Default:

“Please enter your input: “

evofabric.core.agent.input_key#

The key name for user input stored in the state.

Type:

str

Default:

“user_input”

Exception Handling#

UserNode automatically handles the following exceptions:

  • EOFError: User input stream ends (e.g., end-of-file marker)

  • KeyboardInterrupt: User interrupted input (Ctrl+C)

  • Other Exceptions: Catch all unexpected errors

An empty state delta is returned in all exceptional cases to ensure that graph execution is not interrupted.

Contents