Helix Architecture¶

Helix is an enterprise agentic AI platform built in Go that unifies AI agent orchestration, RAG pipelines, and enterprise integration into a single runtime. This document describes the system architecture, core abstractions, and package layout.

Why a Unified Platform?¶

AI agents are only as useful as the data they can access and the systems they can act upon. Most agent frameworks treat enterprise connectivity as an afterthought, requiring teams to bolt on separate integration layers, observability stacks, and security policies. Helix takes a different approach: agents, RAG pipelines, and enterprise integrations share a single runtime with unified lifecycle management, interceptor-based observability (metrics, tracing, audit), production security (JWT, OAuth2, mTLS), graceful shutdown with inflight tracking, and streaming-capable data flow. The result is an AI platform that is production-ready from day one -- agents have reliable access to enterprise data sources, every exchange is observable, and operations teams get the same tooling whether they are managing an AI workflow or a traditional integration route.

Table of Contents¶

• System Overview
• Core Abstractions
• Exchange and Message
• Component
• Endpoint
• Producer and Consumer
• Processor
• Interceptor
• How Workflows Work
• 1. Definition Loading
• 2. URI Resolution
• 3. Pipeline Construction
• 4. Consumer Starts
• 5. Inflight Tracking
• Workflow Lifecycle
• Package Layout
• Agent Framework
• RAG Pipeline
• Protocol Components
• Core Runtime
• Infrastructure
• Components (47 packages in pkg/component/)
• Connector SDK (pkg/connector/)
• CLI
• Subsystem Deep Dives
• Agent Framework
• RAG Pipeline
• Integration Engine
• Connector SDK
• Threading Model
• Streaming Body Model
• Dependency Philosophy

System Overview¶

                       +---------------------------+
                       |       helix CLI            |
                       |  (cmd/helix/main.go)       |
                       +---------------------------+
                                  |
             +-----------+--------+---------+-----------+
             |           |        |         |           |
         +-------+  +-------+ +------+ +--------+ +----------+
         | DSL   |  |Config | |Health| |Metrics | |Management|
         |Loader |  |Loader | |Mgr   | |Collect | |REST API  |
         +-------+  +-------+ +------+ +--------+ +----------+
             |
             v
      +-------------+
      |   Reifier   |  (converts definitions --> runtime workflows)
      +-------------+
             |
             v
  +-------------------+
  |   HelixContext     |  (central runtime: components, workflows, interceptors)
  +-------------------+
        |         |
   +--------+ +--------+
   |Routes  | |Comps   |
   +--------+ +--------+
        |         |
   +---------+  +----------+
   |Processor|  |Endpoint  |
   |Pipeline |  |Producer  |
   +---------+  |Consumer  |
                +----------+

AI agents, RAG pipelines, and enterprise integrations all flow through this same runtime. An agent step in a workflow is just another Processor in the pipeline -- it receives an Exchange, calls an LLM, and passes the result downstream. This means agents automatically inherit the platform's interceptors (metrics, tracing, audit, multi-tenancy), error handling, and inflight tracking without any additional wiring.

Core Abstractions¶

Exchange and Message¶

The Exchange (pkg/exchange/) is the fundamental data envelope flowing through workflows. Every message processed by Helix is wrapped in an Exchange.

type Exchange struct {
    id         string
    pattern    ExchangePattern  // InOnly or InOut
    in         *Message         // inbound message (always present)
    out        *Message         // outbound message (set by processors)
    properties sync.Map         // exchange-scoped metadata
    exception  error            // error state
    unitOfWork *UnitOfWork      // completion callbacks
}

A Message carries a body (any Go value) and a set of headers (string-keyed key-value pairs). Headers use sync.Map for thread safety.

type Message struct {
    body    any
    headers sync.Map
}

Exchange patterns: - InOnly -- fire-and-forget (default) - InOut -- request-reply

Component¶

A Component (pkg/engine/interfaces.go) is a factory for Endpoints. Each component is registered by its URI scheme (e.g., kafka, http, timer). There are 30 built-in components plus the connector meta-component.

type Component interface {
    Scheme() string
    CreateEndpoint(uri string, remaining string, params map[string]string) (Endpoint, error)
    Start(ctx context.Context) error
    Stop(ctx context.Context) error
}

Endpoint¶

An Endpoint represents a specific communication channel created by a component. Given a URI like kafka://orders?brokers=host:9092&group=mygroup, the Kafka component creates an endpoint bound to the orders topic.

type Endpoint interface {
    URI() string
    Component() Component
    CreateProducer() (Producer, error)
    CreateConsumer(processor Processor) (Consumer, error)
}

Producer and Consumer¶

• A Producer sends messages TO an endpoint. It implements the Processor interface so it can be placed in the processing pipeline.
• A Consumer receives messages FROM an endpoint and dispatches them to a processor chain.
• A PollingConsumer extends Consumer for pull-based protocols.

type Producer interface {
    Processor
    Start(ctx context.Context) error
    Stop(ctx context.Context) error
    Endpoint() Endpoint
}

type Consumer interface {
    Start(ctx context.Context) error
    Stop(ctx context.Context) error
    Endpoint() Endpoint
}

Processor¶

A Processor is the fundamental unit of routing logic. Every step in a workflow -- transform, filter, choice, log, setHeader -- is a Processor.

type Processor interface {
    Process(ctx context.Context, exchange *exchange.Exchange) error
}

The ProcessorFunc adapter allows inline functions:

type ProcessorFunc func(ctx context.Context, exchange *exchange.Exchange) error

Interceptor¶

Interceptors provide cross-cutting concerns that wrap every exchange:

type Interceptor interface {
    Before(ctx context.Context, exchange *exchange.Exchange) error
    After(ctx context.Context, exchange *exchange.Exchange) error
}

Built-in interceptors: MetricsInterceptor, TracingInterceptor, AuditInterceptor, TenantInterceptor.

How Workflows Work¶

1. Definition Loading¶

Definitions are written in YAML or JSON and loaded by the DSL package (pkg/dsl/yaml_loader.go). The YAML DSL supports five top-level primitives:

agents:       # AI agents with trigger, model, tools, memory, approval
tools:        # Self-describing tools with mandatory input/output schemas
pipelines:    # Data processing flows (RAG, ETL) with source, steps, output
mcps:         # MCP servers that expose agents/pipelines/workflows as tools
workflows:       # Traditional integration routes (from/steps/to)

The agents:, tools:, and pipelines: primitives are reified into workflows internally -- they provide a simplified, AI-first syntax that maps to the same runtime abstractions. The tools: primitive is special: it requires mandatory description, input, and output schema fields (the parser rejects tools missing any of them), auto-generates a tool://{id} URI (no from: needed), validates input before execution, and exposes full input+output schemas to MCP for discovery. The output schema is stored in the x-helix-output-schema extension. The mcps: primitive configures MCP server instances that expose selected agents, tools, pipelines, and workflows as MCP tools.

Each workflow definition specifies: - A from URI (the consumer endpoint) - A sequence of steps (processors) - Optional error handlers

2. URI Resolution¶

When a workflow starts, the engine parses the from URI using ParseURI:

scheme://remaining?param1=val1&param2=val2

The scheme is looked up in the HelixContext's component registry to find the correct Component, which creates an Endpoint.

3. Pipeline Construction¶

The Reifier (pkg/dsl/reifier.go) converts step definitions into Processor instances. These are chained into a pipeline by BuildPipeline, which wraps them with interceptors and error handling:

Interceptor.Before -> Processor[0] -> Processor[1] -> ... -> Processor[N] -> Interceptor.After

If an exchange is marked as failed (via SetException), subsequent processors are skipped.

4. Consumer Starts¶

The consumer is created from the from endpoint and started. When it receives a message, it creates an Exchange and passes it through the pipeline.

5. Inflight Tracking¶

Exchanges are tracked by the InflightRepository for graceful shutdown. During shutdown, the engine waits up to drainTimeout for all in-flight exchanges to complete.

Workflow Lifecycle¶

Routes pass through these lifecycle states:

New -> Starting -> Started -> (Suspending -> Suspended -> Started) -> Stopping -> Stopped

Routes can be dynamically started, stopped, suspended, resumed, and replaced at runtime through the management API.

Package Layout¶

Agent Framework¶

RAG Pipeline¶

Protocol Components¶

Core Runtime¶

Infrastructure¶

Components (47 packages in pkg/component/)¶

Each component lives in its own package under pkg/component/:

a2a, a2ui, agent, agui, amqp, anthropic, azure, bedrock, cohere, cron, deepinfra, direct, docker, elasticsearch, file, fireworks, ftp, governance, groq, grpc, http, huggingface, kafka, llmutil, log, mcp, milvus, mistral, mock, mongodb, nats, ollama, openai, pinecone, postgres, qdrant, redis, replicate, s3, seda, slack, smtp, sqs, timer, together, tool, vertex, vllm, websocket, xai

Connector SDK (pkg/connector/)¶

The connector SDK provides a framework for building lightweight integrations:

CLI¶

Subsystem Deep Dives¶

Agent Framework¶

The agent framework (pkg/agent/) provides seven built-in agent types that can be used as workflow steps or standalone, plus a RegisterAgentType(name, AgentRunFunc) hook for custom orchestration patterns:

• ReAct - reason-act loop with tool use
• Supervisor - delegates to sub-agents (exposes them as delegate_to_<id> tools)
• Plan-Execute - creates a plan, then executes steps
• Deep - recursive task decomposition (configurable max depth)
• Sequential - runs agents in order
• Parallel - runs agents concurrently (parallel_error_policy: fail_fast / collect / default)
• Loop - iterative refinement

Governance gate. When an AgentConfig.Governance *governance.Stack is wired, every tool dispatch goes through Stack.Check(ctx, agentID, toolName, args) in the order kill switches → privilege rings → policy engine → audit. The first non-allow decision wins; a synthetic Tool call denied by governance: <reason> tool_result is fed back to the LLM, and a governance_block AgentEvent is emitted on the unified OnEvent stream. See the Governance Guide.

Observability. OnEvent is the stable consumer-facing projection — one event per step (model_request / model_response / tool_call / tool_result / approval / guardrail / governance_block / interrupt / context_overflow_retry / …) with DurationMS measured in milliseconds. OnDebugEvent (gated by Debug: true) carries the richer internal DebugEvent for IDE/dev tooling.

Durable interrupt/resume. adk.Runner accepts a CheckpointStore (default: in-process; production: KVCheckpointStore over Redis / memx / SQL). A tool returning tool.Interrupt(info) aborts the run cleanly; Runner.Resume(runID, response) picks up where it left off.

Agents are defined in YAML via the inline agent step with full configuration: model settings (provider, temperature, top_p, stop sequences, response format), tools (including MCP server integration), 8 memory backends (conversation, sliding-window, summary, token-window, long-term, sql-long-term, vector, composite), nested sub-agents, and configurable deep recursion depth. They integrate with LLM providers (OpenAI, Anthropic, Ollama) for inference. See the AI Agents Guide for complete details.

RAG Pipeline¶

The RAG pipeline (pkg/rag/) follows a standard six-stage architecture:

ingest --> chunk --> embed --> store --> retrieve --> generate

Each stage is available as a workflow step: - Ingest -- parse documents (PDF, HTML, Markdown, DOCX, text, CSV, JSON, XML, Starlang -- 10 parsers) - Chunk -- split into segments (fixed, sentence, semantic, recursive, parent-child, token -- 6 strategies) - Embed -- generate vector embeddings (OpenAI, Ollama, Voyage/Anthropic, Cohere -- 4 providers) - Store -- persist in vector databases (Milvus, Qdrant, Pinecone) - Retrieve -- similarity search with optional reranking (MMR, cross-encoder) - Generate -- augment prompts with retrieved context

Integration Engine¶

The integration engine provides the production runtime that AI agents and RAG pipelines depend on. It manages:

1. Component registry -- maps URI schemes to Component instances
2. Workflow management -- add, remove, replace, start, stop, suspend, resume workflows
3. Pipeline execution -- chains processors with interceptor wrapping and error handling
4. Event system -- emits lifecycle events (context starting/started/stopping/stopped, workflow starting/started/stopping/stopped)
5. Inflight tracking -- tracks active exchanges for graceful shutdown drain

The engine uses no external dependencies beyond the Go standard library.

Connector SDK¶

The Connector SDK (pkg/connector/) provides a higher-level abstraction than components for building integrations. Key differences from components:

• Schema-driven configuration with validation and default injection
• Lifecycle management with automatic reconnect and health monitoring
• Middleware chain for retry, circuit breaking, rate limiting, metrics
• Catalog metadata for discovery and UI generation

Connectors are wired into workflows via the connector://name URI scheme through the ConnectorMeta meta-component.

Threading Model¶

• Each workflow consumer runs in its own goroutine
• Agent steps spawn their own goroutines for LLM inference and tool execution; streaming responses use channel-based goroutines that feed tokens back into the exchange pipeline
• SEDA queues use Go channels for async handoff
• Components manage their own connection pooling (Kafka writers, Redis clients, etc.)
• Interceptors and processors run synchronously within the consumer goroutine
• The InflightRepository uses atomic counters and sync.WaitGroup for drain coordination
• All shared state uses sync.RWMutex or sync.Map for thread safety

Streaming Body Model¶

For large payload processing, Helix provides the StreamBody abstraction (pkg/exchange/stream.go) that allows multi-gigabyte data to flow through workflows with constant memory usage.

A StreamBody wraps any io.Reader with metadata (size, MIME type, filename) and is set as the exchange body. Components detect StreamBody automatically and switch to streaming I/O:

• File component (pkg/component/file/streaming.go): StreamingFileReader opens files via os.Open (not os.ReadFile), returning a StreamBody. StreamingFileProducer writes via io.Copy.
• S3 component (pkg/component/s3/multipart.go): MultipartUploader splits large objects into parts uploaded concurrently, with automatic abort on failure. Threshold-based auto-detection (default 100MB).
• HTTP component (pkg/component/http/streaming.go): StreamingHTTPProducer uses chunked transfer encoding for unknown-size bodies. Responses are returned as StreamBody.
• JSON processor (pkg/processor/streaming_json.go): StreamingJSONProcessor uses json.Decoder to process JSON arrays element-by-element without materializing the entire array.

Supporting types include LimitedStreamBody (caps maximum bytes read) and ChunkedReader (fixed-size chunk iteration with offset tracking). Exchange helpers SetStreamBody(), GetStreamBody(), and IsStreamBody() provide a uniform API for stream detection and extraction.

memx Integration Architecture¶

Helix provides native integration with memx, a high-performance vector database that exposes two wire-compatible interfaces. The integration uses dual-mode connectivity with automatic detection based on the connection URL scheme.

Dual-Mode Access¶

Auto-Detection¶

The memx component inspects the connection URL to select the appropriate client:

• http://localhost:8080 or https://memx.prod:8080 -- routes through the memx Go SDK (ES API mode)
• postgres://localhost:5433/memx or postgresql://memx.prod:5433/memx -- routes through pgx/v5 (SQL mode)

No configuration flag is needed; the URL scheme determines the transport.

Architecture Diagram¶

                   +--------------------------+
                   |   Helix Workflow / Agent    |
                   +--------------------------+
                              |
                   +----------+----------+
                   |                     |
            +------+------+      +------+------+
            | memx-go SDK |      |   pgx/v5    |
            | (ES API)    |      |   (SQL)     |
            +------+------+      +------+------+
                   |                     |
              HTTP :8080            pgwire :5433
                   |                     |
                   +----------+----------+
                              |
                   +----------+----------+
                   |     memx server     |
                   |  (vector + hybrid)  |
                   +---------------------+

Column Auto-Discovery¶

When connecting to an existing memx table, Helix auto-detects column roles from the table's index metadata at startup. This eliminates the need for manual id_column, embedding_column, and content_column configuration:

                  +---------------------------+
                  |   memx table indexes      |
                  +---------------------------+
                  | HNSW / IVF / DiskANN idx  |  --> embedding_column
                  | FULLTEXT idx              |  --> content_column
                  | _id suffix / EMBEDDING    |  --> id_column
                  +---------------------------+
                              |
                    +---------+---------+
                    |  Auto-Discovery   |
                    |  (at startup)     |
                    +---------+---------+
                              |
                  +-----------+-----------+
                  |                       |
           +------+------+        +------+------+
           |  rag-store  |        | rag-retrieve|
           | (no column  |        | (no column  |
           |  config)    |        |  config)    |
           +-------------+        +-------------+

The discovery process: 1. Reads table schema and index definitions via the memx API 2. Identifies HNSW, IVF, or DiskANN indexes to find the embedding column 3. Identifies FULLTEXT indexes to find the content column for BM25 search 4. Falls back to column name heuristics (_id suffix, EMBEDDING type) if no indexes match 5. Explicit column overrides (id_column, embedding_column, content_column) take precedence when provided

database: Field¶

Both rag-store and rag-retrieve support a database: field for targeting a specific memx database. This is separate from the connection URL and allows routing to different databases on the same memx instance.

Raw Results Architecture¶

memx retrieve returns raw []map[string]any directly from the table -- all columns are included except the embedding column, with a _score field added. There is no intermediate Chunk wrapping. This design: - Preserves the full table schema in results - Lets agents reason over all available columns - Defers reshaping to downstream Starlang transforms when needed

Integration Points¶

• RAG store/retrieve (rag-store, rag-retrieve): Uses ES API mode for vector insert and similarity search. Supports database: field and auto-discovery.
• Agent vector memory: Stores and recalls conversation embeddings via ES API.
• memx:// component: Direct access to memx for SQL queries (joins, aggregations) or ES API operations from within workflows.
• Hybrid search: Combines dense vector similarity with BM25 keyword scoring via the ES API fusion endpoint.

Multi-Protocol Architecture¶

Helix supports 5 communication protocols, each agent can expose multiple:

                    SynaptX Gateway (:7080)
    ┌──────────────────────────────────────────────┐
    │  /ai/*          → 10 LLM providers            │
    │  /a2a/          → A2A server                   │
    │  /agui/         → AG-UI server                 │
    │  /mcp/          → MCP public (3 tools)         │
    │  /mcp-admin/    → MCP admin (2 tools)          │
    │  /ws/           → WebSocket AI chat            │
    │                                                │
    │  Policies: prompt_guard, PII, WAF, governance  │
    │  Virtual Keys: per-agent budget + rate limit   │
    └──────────────────────────────────────────────┘
                           │
    ┌──────────────────────┼──────────────────────┐
    │                      │                      │
  Helix Agents           MCP Servers           WebSocket
  ├─ :4100 AG-UI (code)  ├─ :3100 (public)    ├─ :8200 (AI chat)
  ├─ :4101 AG-UI (math)  └─ :3101 (admin)     └─ :8201 (echo)
  ├─ :4102 AG-UI (sum)
  ├─ :9100 A2A (code)
  ├─ :9101 A2A (math)
  ├─ :9102 A2A (sum)
  ├─ :9103 A2A (admin)
  └─ :8100 HTTP Chat

Per-Agent Multi-Trigger¶

Each agent can expose multiple protocol endpoints via triggers::

agents:
  - id: code-assistant
    triggers:
      - agui://0.0.0.0:4100/agent    # AG-UI (frontend SSE)
      - a2a-server://0.0.0.0:9100    # A2A (agent-to-agent)
      - http-server://0.0.0.0:8100/api/chat  # HTTP REST

Each trigger creates a separate workflow. The same agent processor pipeline runs regardless of protocol.

Multi-MCP Servers¶

Multiple MCP servers run on separate ports with independent security:

mcps:
  - id: public-tools      # Port 3100 — api_key: public
    expose: [code-assistant, math-agent, summarizer]
  - id: admin-tools        # Port 3101 — api_key: admin
    expose: [admin-tool, code-assistant]
    blocked_tools: [summarizer]

WebSocket Endpoints¶

workflows:
  - id: ws-chat
    trigger: { uri: ws-server://0.0.0.0:8200/ws }
    steps:
      - agent: { id: ws-agent, type: react, ... }

WebSocket server accepts RFC 6455 connections, runs messages through the pipeline, returns the result as a WS frame. Supports text/binary, concurrent clients, and multi-turn conversations.

SynaptX Integration¶

When fronted by SynaptX gateway:

Dependency Philosophy¶

• Core engine: zero external dependencies (pure stdlib)
• Components bring their own deps: kafka-go, go-redis/v9, pgx/v5, mongo-driver/v2, nats.go, amqp091-go
• Expression engine: Starlang (local replace directive)
• Configuration: gopkg.in/yaml.v3
• Security: stdlib crypto + golang.org/x/crypto/bcrypt
• Connectors: stdlib net/http only (no external SDKs)