OpenAgentFlow

Adapters Component

This page documents the runtime adapter module (adapters/), which transforms the Intermediate Representation into executable code for specific runtime frameworks.

Overview

Currently, OAF ships with one adapter:

Adapter Directory Target Framework
LangGraph adapters/langgraph/ LangGraph (Python)

Future planned adapters include AutoGen and CrewAI.

File Class/Export Purpose
adapters/base-adapter.js BaseAdapter Shared, target-agnostic generate() contract: compatibility check → input validation → render
adapters/template-engine.js render() Zero-dependency renderer that substitutes `` placeholders in a stub file
adapters/lang/python.js Type/literal/casing helpers Python formatting helpers shared by any Python-targeting adapter
adapters/langgraph/index.js LangGraphAdapter Maps IR onto template tokens for the LangGraph target
adapters/langgraph/templates/*.py workflow.py, agent_node.py, route_fn.py, required_guard.py Python stub files — the actual target-language source

Static target-language source lives in the stub files under templates/; the .js files read the IR and produce a token map for the template engine to substitute into a stub. The .js files do emit small fragments of Python themselves: the intended case is a per-item expression that can’t be represented as plain data (a lowered when condition, a type, a literal). Two call sites are a deliberate, ruled-on exception where a full per-item statement is built in JS instead — see Stub Files and the Template Engine for the full rule and why.

LangGraph Adapter

The LangGraphAdapter class transforms OAF IR into a self-contained, executable LangGraph Python script.

API

import { LangGraphAdapter } from './adapters/langgraph/index.js';

const adapter = new LangGraphAdapter(ir, options);
const pythonCode = adapter.generate(); // Returns string

Constructor:

Parameter Type Default Description
ir object — The OAF IR document
options object {} Adapter options
options.input object undefined Initial state values from --input JSON file

Methods

generate()

Generates the complete Python source code.

const adapter = new LangGraphAdapter(ir, { input: { feedback: "Great product!" } });
const code = adapter.generate();
// code is a complete Python script ready to execute

checkCompatibility()

Validates that the IR can be compiled to LangGraph.

const compat = adapter.checkCompatibility();
if (!compat.supported) {
    console.error('Issues:', compat.issues.join('; '));
}

Compatibility checks:

Input Validation

When options.input is provided, the adapter validates it against the workflow state:

Validation Error Message
Unknown key Input JSON contains variable "X" which is not defined in workflow state
Type mismatch Type mismatch for state variable "X": expected string, found number
Missing required Missing required initial state variable: "X"

Type compatibility rules:

IR Type Accepted JS Types
string typeof val === 'string'
int typeof val === 'number' && Number.isInteger(val)
float typeof val === 'number'
bool typeof val === 'boolean'
list<*> Array.isArray(val)
map<*,*> typeof val === 'object' && !Array.isArray(val)

Generated Python Structure

The adapter generates a Python script with these sections:

"""
OpenAgentFlow — Generated LangGraph Workflow
Workflow: Customer Feedback Analysis
Generated by: OpenAgentFlow Compiler v0.1.0
"""

# 1. Imports
import os, sys, json
from typing import TypedDict, Optional, List
from langgraph.graph import StateGraph, END
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_openai import ChatOpenAI

# 2. State Schema
class WorkflowState(TypedDict, total=False):
    feedback: Optional[str]    # @required
    sentiment: Optional[str]
    category: Optional[str]
    key_issues: Optional[List[str]]
    response_draft: Optional[str]

# 3. LLM Helper
def get_llm(model=None, temperature=0.7, provider=None):
    """Auto-detects provider from API keys or uses explicit provider."""
    ...

# 4. Agent Node Functions
def sentiment_analyzer_node(state: WorkflowState) -> WorkflowState:
    """Agent: SentimentAnalyzer"""
    ...

def categorizer_node(state: WorkflowState) -> WorkflowState:
    """Agent: Categorizer"""
    ...

# 5. Graph Construction
def build_graph() -> StateGraph:
    graph = StateGraph(WorkflowState)
    graph.add_node("SentimentAnalyzer", sentiment_analyzer_node)
    graph.add_node("Categorizer", categorizer_node)
    graph.set_entry_point("SentimentAnalyzer")
    graph.add_edge("SentimentAnalyzer", "Categorizer")
    graph.add_edge("Categorizer", END)
    return graph.compile()

# 6. Execution
if __name__ == "__main__":
    app = build_graph()
    initial_state = { ... }
    result = app.invoke(initial_state)

Type Mapping

The adapter maps OAF/IR types to Python typing annotations:

OAF Type IR Descriptor Python Type
string "string" str
int "int" int
float "float" float
bool "bool" bool
list[string] "list<string>" List[str]
list[list[int]] "list<list<int>>" List[List[int]]
map[string, int] "map<string,int>" Dict[str, int]
Unknown — Any

All state fields are wrapped in Optional[T] in the generated TypedDict. Fields with @reducer are wrapped in Annotated[Optional[T], operator.add].

Agent Node Generation

Each agent becomes a Python function that:

  1. Gets an LLM instance via get_llm(model, temperature, provider)
  2. Builds a prompt from the agent’s inputs
  3. Calls the LLM via llm.invoke(messages)
  4. Returns state updates based on the outputs

Single Output Agent

When an agent has exactly one output, the LLM response is assigned directly:

def analyst_node(state):
    # ... call LLM ...
    return {"key_points": result}

Multi-Output Agent

When an agent has multiple outputs, the adapter generates JSON parsing logic:

def categorizer_node(state):
    # ... call LLM ...
    try:
        parsed = json.loads(result)
        updates = {}
        if "category" in parsed:
            updates["category"] = parsed["category"]
        if "key_issues" in parsed:
            updates["key_issues"] = parsed["key_issues"]
        return updates
    except (json.JSONDecodeError, TypeError):
        # Fallback: assign raw result to first output
        return {"category": result}

No Output Agent

When an agent has no outputs, it returns an empty dict:

def logger_node(state):
    # ... call LLM ...
    return {}

LLM Provider System

The generated get_llm() function handles provider selection:

graph TD
    A["Agent calls get_llm(model, temp, provider)"] --> B{"Explicit provider?"}
    B -->|"Yes"| C["Use specified provider"]
    B -->|"No"| D{"GOOGLE_API_KEY set?"}
    D -->|"Yes"| E["Use Gemini<br/>ChatGoogleGenerativeAI"]
    D -->|"No"| F{"OPENAI_API_KEY set?"}
    F -->|"Yes"| G["Use OpenAI<br/>ChatOpenAI"]
    F -->|"No"| H["RuntimeError"]

Model Resolution

  1. If the agent has a model property → use that model directly
  2. If no model → check OAF_DEFAULT_MODEL environment variable
  3. If neither → raise a RuntimeError

Import Safety

The generated imports use try/except to gracefully handle missing packages:

_LLM_PROVIDER = None
try:
    from langchain_google_genai import ChatGoogleGenerativeAI
    if os.environ.get("GOOGLE_API_KEY"):
        _LLM_PROVIDER = "gemini"
except ImportError:
    pass

if _LLM_PROVIDER is None:
    try:
        from langchain_openai import ChatOpenAI
        if os.environ.get("OPENAI_API_KEY"):
            _LLM_PROVIDER = "openai"
    except ImportError:
        pass

State Initialization

The generated __main__ block supports state injection from multiple sources:

1. Compile-Time Embedding

When --input data.json is passed to oaf compile, the adapter embeds the values directly:

initial_state = {
    "feedback": "Great product but needs improvements",
    "sentiment": "",
    "category": "",
}

2. Runtime Override

The generated script also reads --input or OAF_INPUT_FILE at runtime:

input_file = os.environ.get("OAF_INPUT_FILE")
for idx in range(len(args)):
    if args[idx] in ("--input", "-i") and idx + 1 < len(args):
        input_file = args[idx + 1]
        break

if input_file:
    with open(input_file, "r", encoding="utf-8") as f:
        runtime_input = json.load(f)
        if isinstance(runtime_input, dict):
            initial_state.update(runtime_input)

3. Required Fields Validation

If state variables have @required, the generated script validates them:

missing_required = [
    f for f in ["feedback"]
    if initial_state.get(f) is None or 
       (isinstance(initial_state.get(f), str) and initial_state.get(f) == "")
]
if missing_required:
    print(f"Error: Missing required state variables: {', '.join(missing_required)}")
    sys.exit(1)

Console Encoding Safety

The generated script includes encoding safety for Windows terminals:

if hasattr(sys.stdout, 'reconfigure'):
    sys.stdout.reconfigure(encoding='utf-8')

This prevents crashes on Windows terminals with non-UTF-8 codepages (e.g., cp1256, cp1252).

Stub Files and the Template Engine

Generated Python does not come from JavaScript string literals. It comes from stub files — plain .py files under adapters/langgraph/templates/ that are valid-looking Python containing `` placeholders — rendered by the zero-dependency engine in adapters/template-engine.js.

Stub Purpose
workflow.py The main stub: file header, imports, WorkflowState TypedDict, get_llm() helper, graph construction, and the __main__ execution block
agent_node.py One agent node function, rendered once per agent
route_fn.py One conditional-routing function, rendered once per group of edges sharing a source where any edge has a when condition
required_guard.py The runtime @required state-variable guard, rendered only when the workflow declares required fields

render(stubPath, tokens) reads a stub and substitutes each `` placeholder with the matching value from the token map. Two substitution modes, chosen by how the placeholder appears in the stub:

The engine throws if a stub declares a placeholder with no matching token, or if a token has no matching placeholder — this is what catches drift when a stub is edited without updating the adapter that feeds it, or vice versa.

The LangGraphAdapter.buildTokens() method (adapters/langgraph/index.js) is where IR inspection, type conversion, and structure-building happen; it returns a plain token map and has no knowledge of how the engine applies it. This keeps compiler logic and target-language emission mostly separate, under one rule with two named exceptions:

The rule: JS may emit target-language source only when the content is a per-item expression that cannot exist as data — a lowered when condition (route_fn.py’s BRANCHES, built with irToPythonExpr), a type, a literal. A per-item statement over a list of names should be a JSON data token plus a static loop written once in the stub — the pattern OUTPUTS (agent_node.py) and REQUIRED_FIELDS (required_guard.py) both use: index.js hands over JSON.stringify(...), and the stub’s own Python does the iterating at runtime. Static boilerplate always lives in a stub.

Ruled-on exception, kept on purpose: GRAPH_NODES’s graph.add_node(...) calls and _graphEdges’s graph.add_edge(...) lines (index.js:57-59,144) are built as full statements in JS rather than data-driven. This could be data-driven, but the generated Python is meant to be read and edited by users, and an explicit call per node/edge reads better than a loop over an opaque list — a deliberate trade-off from the final branch review, not a gap to close.

Known gap, deferred: _userMessage’s per-input if state.get(...): user_parts.append(...) pair (index.js:85-96) is the kind of per-item statement the rule says shouldn’t live in JS — the fix is one static list comprehension in agent_node.py plus an `` data token. Not fixed on this branch because closing it changes every agent’s generated bytes, which needs a deliberate, reviewed snapshot re-record — tracked as follow-up work, not silently left broken.

Writing a New Adapter

Adding a target framework (AutoGen, CrewAI, or any other) follows the same shape as the LangGraph adapter:

  1. Extend BaseAdapter (adapters/base-adapter.js). It owns the shared generate() contract — compatibility check, then input validation, then render — so a new adapter only needs to supply the pieces specific to its target.
  2. Define three members on the subclass:
    • templateDir — the absolute path to the adapter’s stub directory (typically fileURLToPath(new URL('./templates/', import.meta.url))).
    • mainTemplate — the stub filename rendered as the top-level output document.
    • buildTokens() — reads this.ir and returns a Record<string, string | string[]> token map. This is the only place the new adapter inspects the IR; array values are joined with newlines by the template engine.
  3. Put target-language source in stub files under adapters/<target>/templates/. Keep the file extension of the target language (.py, .ts, .js, …) so editors give correct syntax highlighting; embed `` placeholders wherever buildTokens() needs to inject content. The adapter’s .js files should build target-language source only as a small per-item expression that can’t be represented as data — see Stub Files and the Template Engine for the full rule, including its two named, reviewed exceptions.
  4. Register the target in the --target switch in cli/index.js, alongside the existing ir and langgraph cases, so oaf compile --target <name> and (if the target is executable) oaf run --target <name> can reach it.
  5. If the new adapter targets Python, reuse adapters/lang/python.js for type mapping, literal rendering, and identifier casing rather than duplicating those helpers — they are already framework-agnostic.

Complete Example

import { Compiler } from './compiler/index.js';
import { LangGraphAdapter } from './adapters/langgraph/index.js';
import { readFileSync, writeFileSync } from 'fs';

// 1. Compile the .oaf file
const source = readFileSync('examples/summarize.oaf', 'utf-8');
const compiler = new Compiler(source, 'summarize.oaf');
const result = compiler.compile();

if (result.status !== 'success') {
    console.error('Compilation failed');
    process.exit(1);
}

// 2. Load input data
const inputData = JSON.parse(readFileSync('examples/summarize-input.json', 'utf-8'));

// 3. Generate Python code
const adapter = new LangGraphAdapter(result.ir, { input: inputData });
const pythonCode = adapter.generate();

// 4. Save to file
writeFileSync('summarize.py', pythonCode, 'utf-8');
console.log('Generated summarize.py');

Next Steps

This site is open source. Improve this page.