Custom Python Nodes — Write Deterministic Code in AI Graphs

What is a Custom Python Node?

While Large Language Models (LLMs) are excellent at understanding intent and generating conversation, they are famously unreliable at deterministic calculations — performing exact mathematics, parsing structured formats, or transforming complex arrays.

To bridge this gap, Langoedge provides a Custom Python Node. It allows you to run pure Python script directly within your agent's graph to execute exact logic with 100% predictability.

Key Concept: A Custom Python Node is the "calculator" of your AI agent's mind. Use it whenever you need strict calculations, data formatting, or validation before proceeding.

How It Works: The Secure Sandbox

For enterprise security and reliability, Langoedge executes custom Python code inside an isolated execution sandbox based on Python 3.11+.

1. Syntax Parsing: When the graph is run, the backend compiles and validates your Python code to ensure compliance with our security standards.
2. Safety Audits: The compiler checks every statement against a strict whitelist. If it detects unapproved operations, execution halts instantly.
3. Namespace Isolation: The code is run in an isolated local execution scope, preventing side effects or interference across threads.

State Interoperability: Function Structure

Every Custom Python Node must define exactly one Python function. Langoedge compiles this function and passes the graph's shared State object (a dictionary) as the first parameter. To update the graph memory, your function must return a dictionary containing the state fields you want to update.

[!IMPORTANT]
Since the execution engine validates and extracts exactly one function definition, all import statements, helper functions, and logic must be contained within the node's code, and the state must be accessed inside the function signature (not as a global variable).

Reading from State

The input state parameter is a dictionary representing the graph's memory. You can read any of the 15 standard state fields (field1 through field15) or lists:

def my_custom_node(state):
    # Accessing the latest user message
    latest_msg = state.get("field1", [])[-1] if state.get("field1") else ""
    
    # Accessing a custom variable (e.g., extracted email)
    extracted_email = state.get("field5", [])[-1] if state.get("field5") else None

Writing to State

To update the graph's memory, return a dictionary from your function. The returned keys must correspond to the state fields, and the values should be lists (since Langoedge uses append-only list reducers for all state fields):

def my_custom_node(state):
    # Return a dictionary with lists of updates to append back to the State fields
    return {
        "field5": ["support@langoedge.com"],  # Appends to field5 list
        "field15": ["Operation Successful"]   # Appends to field15 list
    }

Developer Recipes

Here are three common real-world recipes for Langoedge Custom Python Nodes.

Recipe 1: String Parsing & JSON Validation

AI agents often output structured data wrapped in conversational markdown (e.g., inside ```json blocks). This node extracts the clean JSON string and parses it safely.

def parse_and_validate_refund(state):
    import json

    # Fetch the raw, messy AI text output from field3
    raw_ai_text = state.get("field3", [])[-1] if state.get("field3") else ""

    # Strip markdown block indicators if present
    cleaned = raw_ai_text.replace("```json", "").replace("```", "").strip()
    
    try:
        data = json.loads(cleaned)
        # Validate critical fields
        if "order_id" in data and "refund_amount" in data:
            return {
                "field6": [data["order_id"]],
                "field7": [str(float(data["refund_amount"]))],
                "field8": ["VALIDATED"]
            }
    except Exception:
        pass

    return {
        "field6": [""],
        "field7": ["0.0"],
        "field8": ["INVALID"]
    }

Recipe 2: Smart Date & Time Conversion

Users say things like "tomorrow at 3 PM" or "next Friday". While the LLM can extract relative values, your database or API requires an exact ISO-8601 string. Use Python's built-in datetime library to resolve relative times.

def calculate_target_date(state):
    from datetime import datetime, timedelta

    # Assume the LLM extracted a relative days offset in field4 (e.g., "1" or "7")
    relative_offset_str = state.get("field4", [])[-1] if state.get("field4") else "0"

    try:
        days_offset = int(relative_offset_str)
    except ValueError:
        days_offset = 0

    # Calculate exact calendar date
    now = datetime.now()
    target_date = now + timedelta(days=days_offset)

    # Format to database-friendly ISO format (YYYY-MM-DD)
    formatted_iso = target_date.strftime("%Y-%m-%d")

    return {
        "field9": [formatted_iso],
        "field10": [f"Scheduled for {formatted_iso}"]
    }

Recipe 3: Advanced Array Transformations

Filter conversation logs, extract specific key metrics, or clean up lists before passing them to a customer CRM or email provider.

def transform_logs(state):
    # Read the chat history from field1
    chat_history = state.get("field1", [])

    # Extract only human questions (filtering out bot replies)
    human_messages = [msg for msg in chat_history if msg.get("role") == "user"]

    # Take the last 3 human queries and join them for a summaries report
    recent_queries = [msg.get("content", "") for msg in human_messages[-3:]]
    formatted_report = " | ".join(recent_queries)

    return {
        "field12": [formatted_report],
        "field13": [f"Total human turns: {len(human_messages)}"]
    }

Best Practices

Frequently Asked Questions