Documents
Home>Documents>AI>Agent>Xgen

Node-Based Architecture: ChatOpenAI, VectorStore, and Agent Nodes

6 min readMar 20, 2025Feb 22, 2026

Node-Based Architecture Design - ChatOpenAI, VectorStore, and Agent Nodes

The core of the XGen platform is the abstraction of various AI capabilities into discrete units called nodes. This post documents how the node-based architecture was designed and implemented.

Abstract Node Class Design

An abstract base class was defined to enforce a consistent interface across all nodes.

from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from pydantic import BaseModel

class NodeConfig(BaseModel):
    node_id: str
    node_type: str
    params: Dict[str, Any] = {}

class BaseNode(ABC):
    def __init__(self, config: NodeConfig):
        self.config = config
        self.node_id = config.node_id

    @abstractmethod
    async def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        pass

    async def validate(self) -> bool:
        return True

    def get_output_schema(self) -> Dict:
        return {"output": "any"}

Key Node Implementations

ChatOpenAI Node

from openai import AsyncOpenAI

class ChatOpenAINode(BaseNode):
    def __init__(self, config: NodeConfig):
        super().__init__(config)
        self.client = AsyncOpenAI()
        self.model = config.params.get("model", "gpt-4o")
        self.temperature = config.params.get("temperature", 0.7)
        self.system_prompt = config.params.get("system_prompt", "")

    async def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        user_input = inputs.get("input", "")
        messages = []
        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})
        messages.append({"role": "user", "content": user_input})

        response = await self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=self.temperature,
        )
        return {"output": response.choices[0].message.content}

ChatAnthropic Node

from anthropic import AsyncAnthropic

class ChatAnthropicNode(BaseNode):
    def __init__(self, config: NodeConfig):
        super().__init__(config)
        self.client = AsyncAnthropic()
        self.model = config.params.get("model", "claude-sonnet-4-20250514")

    async def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=4096,
            messages=[{"role": "user", "content": inputs.get("input", "")}],
        )
        return {"output": response.content[0].text}

Node Factory Pattern

A factory was implemented to instantiate the appropriate node from a node type string.

NODE_REGISTRY: Dict[str, type] = {
    "ChatOpenAI": ChatOpenAINode,
    "ChatAnthropic": ChatAnthropicNode,
    "VectorStore": VectorStoreNode,
    "Retriever": RetrieverNode,
    "Agent": AgentNode,
    "Embedding": EmbeddingNode,
}

def create_node(config: NodeConfig) -> BaseNode:
    node_cls = NODE_REGISTRY.get(config.node_type)
    if not node_cls:
        raise ValueError(f"Unknown node type: {config.node_type}")
    return node_cls(config)

Designing for Extensibility

Adding a new node only requires subclassing BaseNode and implementing the execute method. This made it straightforward to introduce additional node types — such as OCR and training nodes — as the project progressed. The system eventually grew to support over ten node types, and data flow between nodes was handled naturally by the architecture.

The most important aspect of the node architecture is consistency in input/output schemas. Because every node exchanges Dict[str, Any], any node can be freely connected to any other.

Tags
NodeEditorArchitectureChatOpenAIAgentDesignPattern