Home

AgentGraph Compiler — Developer Wiki

Complete reference for every YAML field, node type, edge type, and generated output. Read this to understand exactly what AgentGraph Compiler compiles and why.

How AgentGraph Compiler Works
YAML Spec — Top-Level Structure
graph — Graph Configuration
State — TypedDict vs Pydantic
nodes — Node Types
chains — LLM Chain Configuration
edges — Edge Types
tools — Tool Configuration
Memory & Checkpointing
Hierarchical Sub-Agents (graphRef)
Generated Output — File by File
CLI Reference
Global Configuration (configs/config.yaml)
Complete End-to-End Example
Roadmap

1. How AgentGraph Compiler Works

               ┌───────────────────────────────┐
               │         agent.yaml            │
               │  (YAML Graph Specification)  │
               └───────────────┬───────────────┘
                               │
                               ▼
               ┌───────────────────────────────┐
               │      agentgraph-compiler      │
               │     (Validation & Code Gen)   │
               └───────────────┬───────────────┘
                               │
               ┌───────────────┼───────────────┐
               ▼               ▼               ▼
        graphs/state.py  graphs/graph.py  graphs/nodes/*

AgentGraph Compiler reads a declarative YAML file and generates a complete, runnable Python codebase wired to LangGraph. You never write StateGraph, add_node, or add_edge by hand.

Compilation Pipeline

Parse YAML → Python dict
Validate schema (Pydantic) → check edge targets, supervisor route fields, fan_out params
Create directories → graphs/, nodes/, chains/, tools/, routers/
Generate graphs/state.py → TypedDict or Pydantic state class
Scaffold nodes → one .py file per node with typed function stub
Scaffold chains → one .py file per chain with LLM binding comments
Scaffold tools → one .py file per tool (or MCP client stub)
Scaffold routers → conditional edge router function stubs
Generate graphs/graph.py → fully-wired StateGraph with all edges, ToolNode, Send fan-outs, checkpointer
Generate main.py → runnable entrypoint with app.stream()
Generate flowchart.md → Mermaid diagram of the compiled graph

2. YAML Spec — Top-Level Structure

Every agent YAML can have the following top-level keys:

version: "1.0"      # Optional. Schema version for forward-compat tracking.

graph:              # Required. Graph metadata and state configuration.
  ...

nodes:              # Required. All nodes in the graph.
  ...

chains:             # Optional. LLM chain / runnable definitions.
  ...

edges:              # Required. Graph topology (connections between nodes).
  ...

tools:              # Optional. LangChain tools or MCP client stubs.
  ...

retriever:          # Optional. Vector store retriever configuration.
  ...

3. `graph` — Graph Configuration

The graph block controls the output directory, state schema, shared fields, and persistence.

graph:
  source_dir: "my_agent"                    # (str) Output directory for all generated Python files.
                                             # Sub-agents are nested under this directory.

  state_type: typeddict                      # (str) "typeddict" (default) | "pydantic"
                                             # Controls what base class state.py generates.

  state_schema: "graphs.state.AgentState"   # (str) Dotted Python path to the state class.
                                             # Format: "<module_path>.<ClassName>"

  state_fields:                             # (dict) Field name → Python type string.
    messages: "List[BaseMessage]"            # → auto-wired with add_messages reducer
    next: "str"                              # Simple scalar field
    context: "dict"                          # Dict field
    documents: "List[str]"                   # List of strings
    score: "float"                           # Numeric field
    approved: "bool"                         # Boolean field

  shared_fields: [messages, context]         # (list) Fields readable/writable by sub-agents.
                                             # Annotated in state.py docstring for clarity.

  memory:                                    # (dict) Checkpointing configuration.
    type: sqlite                             # "sqlite" | "in_memory" | omit for stateless
    db_path: ".checkpoints/agent.db"         # Only used for sqlite.

`state_fields` Type Mapping

YAML type string	Generated TypedDict field	Generated Pydantic field
`"List[BaseMessage]"`	`Annotated[List[BaseMessage], add_messages]`	`Annotated[List[BaseMessage], add_messages] = Field(default_factory=list)`
`"List[str]"`	`List[str]`	`List[str] = Field(default_factory=list)`
`"str"`	`str`	`str = Field(default='')`
`"Optional[str]"`	`Optional[str]`	`Optional[str] = Field(default='')`
`"dict"`	`dict`	`dict = Field(default_factory=dict)`
`"int"` / `"float"`	`int` / `float`	`int = Field(default=0)`
`"bool"`	`bool`	`bool = Field(default=False)`
`"Any"`	`Any`	`Any = Field(...)`

4. State — TypedDict vs Pydantic

TypedDict (Default — LangGraph Native)

graph:
  state_type: typeddict
  state_fields:
    messages: "List[BaseMessage]"
    question: "str"
    generation: "str"

Generated graphs/state.py:

# Auto-generated state schema — TypedDict (LangGraph-native)
from typing import TypedDict, List, Annotated
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    question: str
    generation: str

TypedDict is the preferred LangGraph state type. Annotated[..., add_messages] means each update appends messages rather than overwriting the list — critical for multi-turn conversation history.

Pydantic (Legacy / Advanced Serialization)

graph:
  state_type: pydantic
  state_fields:
    question: "str"
    documents: "List[str]"

Generated graphs/state.py:

# Auto-generated state schema — Pydantic BaseModel
from pydantic import BaseModel, Field
from typing import List

class AgentState(BaseModel):
    question: str = Field(default='', description='Generated field')
    documents: List[str] = Field(default_factory=list, description='Generated field')

Use pydantic when you need custom validators, serializers, or integration with Pydantic-aware systems.

5. `nodes` — Node Types

5.1 Standard Node

The most common type. Calls a Python function that takes state and returns a dict of updates.

nodes:
  retrieve:
    module: "graphs.nodes.retrieve"     # (str) Python module path (dotted).
    function: "retrieve"               # (str) Function name inside the module.
    chains: ["retrieval_grader"]       # (list) Chain names to import into this node.
    description: "Pulls document chunks from the vector store."

Generated graphs/nodes/retrieve.py:

from my_agent.graphs.state import AgentState
from my_agent.graphs.chains import retrieval_grader

def retrieve(state: AgentState) -> dict:
    """
    Pulls document chunks from the vector store.
    """
    print("---RUNNING NODE: retrieve---")

    # Placeholder: implement node logic here and return a state update dict.
    return {}

You implement:

def retrieve(state: AgentState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"documents": [d.page_content for d in docs]}

5.2 Async Node

Use when the node performs async I/O (concurrent HTTP calls, async LLM clients, database queries).

nodes:
  async_retriever:
    module: "graphs.nodes.async_retriever"
    function: "async_retriever"
    async_fn: true                     # ← scaffolds `async def`
    description: "Async concurrent document retrieval."

Generated graphs/nodes/async_retriever.py:

from my_agent.graphs.state import AgentState

async def async_retriever(state: AgentState) -> dict:
    """
    Async concurrent document retrieval.
    """
    # This is an async node — use `await` for async LLM/tool calls.
    print("---RUNNING ASYNC NODE: async_retriever---")
    return {}

You implement:

async def async_retriever(state: AgentState) -> dict:
    results = await asyncio.gather(
        vector_store.asimilarity_search(state["question"]),
        web_client.aget(state["question"]),
    )
    return {"documents": [doc for r in results for doc in r]}

Note: Run async graphs with app.astream() (not app.stream()).

5.3 Supervisor Node

An LLM-driven router that reads the user request and writes a routing decision to state[route_field].

nodes:
  supervisor:
    type: supervisor                   # Required. Triggers supervisor generator.
    provider: openai                   # openai | anthropic | ollama | groq
    llm: gpt-4o                        # Model name for the LLM.
    route_field: next                  # (str) State field to write routing decision to. Default: "next"
    system_prompt: "templates/supervisor.prompt"  # Path to a .prompt template file.
    routes_to: [rag_agent, code_agent] # (list) Node names this supervisor can dispatch to.
    description: "Routes requests to the appropriate sub-agent."

Generated graphs/nodes/supervisor.py:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from my_agent.graphs.state import AgentState

llm = ChatOpenAI(model="gpt-4o", temperature=0)

SYSTEM_PROMPT = """You are a supervisor routing requests to: rag_agent, code_agent.
Respond with JSON: {"next": "<agent_name>"}"""

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    ("human", "{messages}"),
])

chain = prompt | llm

def supervisor(state: AgentState) -> dict:
    """LLM-driven routing supervisor."""
    result = chain.invoke({"messages": state["messages"]})
    # TODO: parse result to extract routing decision
    return {"next": "rag_agent"}  # Replace with parsed result

How routing works:

The supervisor_route edge reads state["next"] (or your custom route_field) to dispatch:

supervisor → state["next"] == "rag_agent"  → rag_agent node
           → state["next"] == "code_agent" → code_agent node
           → state["next"] == "__end__"    → post_process (convergence)

5.4 Sub-Agent Reference (`graphRef`)

References another YAML file. AgentGraph Compiler compiles it recursively and nests the output under the parent's source_dir.

nodes:
  rag_agent:
    graphRef: "sub-agents/adaptive-rag.yaml"  # Relative path from this YAML's directory.
    description: "Adaptive RAG sub-agent for knowledge-base questions."

The sub-agent YAML is compiled independently and its compiled app is imported into the parent graph as a node:

# Auto-generated in parent graphs/graph.py:
from my_agent.adaptive_rag.graphs.graph import app as rag_agent_app
workflow.add_node('rag_agent', rag_agent_app)

See Section 10 for full sub-agent documentation.

5.5 Human Review Node (auto-generated for `interrupt_before`)

When an edge has interrupt_before: true, AgentGraph Compiler scaffolds a human_review node:

# Auto-generated graphs/nodes/human_review.py
from my_agent.graphs.state import AgentState

def human_review(state: AgentState) -> dict:
    """
    Pauses graph execution for human review/approval.
    Resume by calling: app.invoke(None, config={"configurable": {"thread_id": "..."}})
    Or update state first: app.update_state(config, {"approved": True})
    """
    print("---HUMAN REVIEW NODE: awaiting review input---")
    return {}

6. `chains` — LLM Chain Configuration

Chains are LangChain runnables (prompt | llm | parser). AgentGraph Compiler scaffolds the file with the correct provider imports and LLM binding comments.

chains:
  retrieval_grader:
    module: "graphs.chains.retrieval_grader"   # (str) Python module path.
    runnable: "retrieval_grader"               # (str) Variable name exported from the module.
    provider: anthropic                         # openai | anthropic | ollama | groq
    llm: claude-3-5-haiku-20241022             # Model name.
    temperature: 0.0                            # LLM temperature.
    prompt: "retrieval_grader"                  # Load from templates/retrieval_grader.prompt
    tools: [web_search]                         # Tool names to import and bind to LLM.
    description: "Binary relevance grader — returns 'yes' or 'no'."

Generated graphs/chains/retrieval_grader.py:

"""
Binary relevance grader — returns 'yes' or 'no'.
"""
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# --- LLM Configuration (Anthropic) ---
# TODO: Uncomment and configure your preferred LLM provider:
# from langchain_anthropic import ChatAnthropic
# llm = ChatAnthropic(model="claude-3-5-haiku-20241022", temperature=0.0)

# --- Prompt Template ---
prompt = ChatPromptTemplate.from_messages([
    ("system", """<content from templates/retrieval_grader.prompt>"""),
    ("human", "{input}"),
])

# --- Chain Definition ---
# If using an LLM: retrieval_grader = prompt | llm | StrOutputParser()
retrieval_grader = prompt | StrOutputParser()

You implement:

from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import JsonOutputParser

llm = ChatAnthropic(model="claude-3-5-haiku-20241022", temperature=0.0)
retrieval_grader = prompt | llm | JsonOutputParser()

Supported Providers

`provider` value	Import scaffolded	Class
`openai` (default)	`from langchain_openai import ChatOpenAI`	`ChatOpenAI`
`anthropic`	`from langchain_anthropic import ChatAnthropic`	`ChatAnthropic`
`ollama`	`from langchain_ollama import ChatOllama`	`ChatOllama`
`groq`	`from langchain_groq import ChatGroq`	`ChatGroq`

7. `edges` — Edge Types

7.1 `standard` — Direct Edge

A simple A → B connection with no routing logic.

- type: standard
  start: retrieve            # Source node name (or START)
  end: grade_documents       # Destination node name (or END)

Generated:

workflow.add_edge('retrieve', 'grade_documents')

7.2 `conditional` — Router Function Edge

Calls a Python function to decide which node to go to next.

- type: conditional
  start: grade_documents
  router:
    module: "routers.router"              # Module containing the router function.
    function: "route_after_grading"       # Function name.
  mappings:
    web_search: web_search               # return value → node name
    generate: generate

Generated routers/router.py (stub):

from my_agent.graphs.state import AgentState

def route_after_grading(state: AgentState) -> str:
    """
    Conditional edge router. Return one of: ['web_search', 'generate']
    """
    print("---RUNNING ROUTER: route_after_grading---")
    # TODO: implement routing logic
    # Example: return "web_search" if state["web_search"] else "generate"
    return "generate"

Generated in graphs/graph.py:

from my_agent.routers.router import route_after_grading

workflow.add_conditional_edges(
    'grade_documents',
    route_after_grading,
    {'web_search': 'web_search', 'generate': 'generate'}
)

You implement the router:

def route_after_grading(state: AgentState) -> str:
    return "web_search" if state["web_search"] else "generate"

7.3 `supervisor_route` — Supervisor Dispatch

Routes from a supervisor node based on the value of state[route_field]. Also wires return edges from each sub-agent back to the convergence node.

- type: supervisor_route
  start: supervisor
  routes_to: [rag_agent, code_agent]     # Possible dispatch targets.
  end: post_process                       # Convergence node all agents return to.

Generated in graphs/graph.py:

# Supervisor routing: dispatches based on state['next']
workflow.add_conditional_edges(
    'supervisor',
    lambda state: state['next'] if isinstance(state, dict) else getattr(state, 'next'),
    {'rag_agent': 'rag_agent', 'code_agent': 'code_agent', '__end__': 'post_process'}
)
# Return edges: sub-agents → convergence node
workflow.add_edge('rag_agent', 'post_process')
workflow.add_edge('code_agent', 'post_process')

Compile-time validation: AgentGraph Compiler raises ValueError if route_field (default: next) is not declared in state_fields.

7.4 `tool_loop` — ReAct Tool Loop

Wires the ReAct pattern: LLM node → ToolNode → LLM node — looping until no tool calls remain.

- type: tool_loop
  start: llm_agent           # LLM node (must have tool_calls in state or as return value)
  tool_node: tools            # ToolNode name (auto-created by compiler)
  end: END                    # Where to go when no more tool calls

Generated in graphs/graph.py:

from langgraph.prebuilt import ToolNode
from my_agent.graphs.tools import tools   # list of LangChain tool functions

workflow.add_node('tools', ToolNode(tools=tools))

# Tool loop: llm_agent ↔ tools
workflow.add_edge('tools', 'llm_agent')
workflow.add_conditional_edges(
    'llm_agent',
    lambda s: 'tools' if (s.get('tool_calls') if isinstance(s, dict) else getattr(s, 'tool_calls', None)) else '__end__',
    {'tools': 'tools', '__end__': END}
)

Auto-generated graphs/tools/__init__.py:

# Export tools list for ToolNode
from my_agent.graphs.tools.web_search import web_search
from my_agent.graphs.tools.calculator import calculator

tools = [web_search, calculator]
__all__ = ['tools']

7.5 `fan_out` — Parallel Dispatch (Send API)

Dispatches one node's output to multiple instances of another node simultaneously, using LangGraph's Send API.

- type: fan_out
  start: coordinator         # Node that produces the list to fan out over
  target: worker             # Node to run in parallel for each item
  field: tasks               # State field (must be a list) to iterate over
  key: task                  # Key name in each Send payload dict
  end: aggregator            # Convergence node after all workers finish

Generated in graphs/graph.py:

from langgraph.types import Send

# Fan-out: Send each item in state['tasks'] to 'worker' in parallel
def _fan_out_coordinator_to_worker(state):
    items = state['tasks'] if isinstance(state, dict) else getattr(state, 'tasks', [])
    return [Send('worker', {'task': item}) for item in items]

workflow.add_conditional_edges(
    'coordinator',
    _fan_out_coordinator_to_worker,
    ['worker']
)
workflow.add_edge('worker', 'aggregator')

How it works:

coordinator → [Send(worker, {task: "A"}), Send(worker, {task: "B"}), Send(worker, {task: "C"})]
                    ↓ parallel execution ↓
             worker(task="A")  worker(task="B")  worker(task="C")
                    ↓ all complete ↓
                   aggregator

7.6 `interrupt_before` — Human-in-the-Loop

Pauses graph execution before a specified node. Requires a persistent checkpointer (sqlite).

- type: standard
  start: drafter
  end: human_review
  interrupt_before: true     # Pause before 'human_review' runs

Generated in graphs/graph.py:

app = workflow.compile(
    checkpointer=_checkpointer,
    interrupt_before=['human_review']
)

Resume patterns:

thread_config = {"configurable": {"thread_id": "review-thread-1"}}

# Option 1: Resume without changes
for event in app.stream(None, config=thread_config):
    print(event)

# Option 2: Inject feedback then resume
app.update_state(thread_config, {"feedback": "Shorten the intro.", "approved": True})
for event in app.stream(None, config=thread_config):
    print(event)

8. `tools` — Tool Configuration

8.1 Standard LangChain Tool

tools:
  web_search:
    module: "graphs.tools.web_search"
    function: "web_search"
    description: "Searches the web via Tavily for up-to-date information."

Generated graphs/tools/web_search.py:

from langchain_core.tools import tool

@tool
def web_search(query: str) -> str:
    """
    Searches the web via Tavily for up-to-date information.
    """
    print("---RUNNING TOOL: web_search---")
    # Placeholder: return string output
    return "Placeholder response from tool: web_search"

You implement:

from langchain_tavily import TavilySearch

search = TavilySearch(max_results=3)

@tool
def web_search(query: str) -> str:
    """Searches the web via Tavily for up-to-date information."""
    results = search.invoke(query)
    return "\n".join(r["content"] for r in results)

8.2 MCP Tool Client Stub

tools:
  knowledge_base:
    module: "graphs.tools.knowledge_base"
    function: "knowledge_base"
    source: mcp                              # ← triggers MCP stub generation
    server_url: "http://localhost:8000/mcp"
    description: "Queries internal knowledge base via MCP server."

Generated graphs/tools/knowledge_base.py:

"""
MCP Tool Client scaffold for knowledge_base.
Connects to external MCP Server at http://localhost:8000/mcp.
"""
from langchain_core.tools import tool

@tool
def knowledge_base(query: str) -> str:
    """
    Queries internal knowledge base via MCP server. (MCP Tool Client)
    """
    print("---RUNNING MCP TOOL: knowledge_base -> http://localhost:8000/mcp---")
    # TODO: Initialize MCP Client session and call remote tool
    return f"MCP Tool response from http://localhost:8000/mcp for: {query}"

9. Memory & Checkpointing

SQLite (Persistent — recommended for production)

graph:
  memory:
    type: sqlite
    db_path: ".checkpoints/agent.db"

Generated in graphs/graph.py:

from langgraph.checkpoint.sqlite import SqliteSaver
import os

_DB_PATH = ".checkpoints/agent.db"
os.makedirs(os.path.dirname(_DB_PATH), exist_ok=True)
_checkpointer = SqliteSaver.from_conn_string(_DB_PATH)

app = workflow.compile(checkpointer=_checkpointer)

In-Memory (Ephemeral — good for development/testing)

graph:
  memory:
    type: in_memory

Generated:

from langgraph.checkpoint.memory import MemorySaver
_checkpointer = MemorySaver()
app = workflow.compile(checkpointer=_checkpointer)

Stateless (No Checkpointer)

Omit the memory block entirely:

graph:
  source_dir: "my_agent"
  # no memory: block

Generated:

_checkpointer = None
app = workflow.compile()

10. Hierarchical Sub-Agents (`graphRef`)

Sub-agents allow you to compose complex multi-agent systems from modular, independently-testable sub-graphs.

How It Works

configs/agents/
├── agent.yaml                 ← parent: references sub-agents
└── sub-agents/
    ├── rag_agent.yaml         ← compiled independently
    └── code_agent.yaml        ← compiled independently

Parent YAML:

nodes:
  rag_agent:
    graphRef: "sub-agents/rag_agent.yaml"    # Path relative to THIS yaml file
    description: "RAG sub-agent."

What happens during compilation:

rag_agent.yaml is compiled first, generating code in my_agent/rag_agent/
The parent's graphs/graph.py imports the sub-agent's compiled app:

from my_agent.rag_agent.graphs.graph import app as rag_agent_app
workflow.add_node('rag_agent', rag_agent_app)

Output Directory Layout

my_agent/                    ← parent source_dir
├── graphs/graph.py           ← parent graph (imports sub-agent apps)
├── graphs/state.py           ← parent state
└── rag_agent/               ← sub-agent nested here (not at root!)
    ├── graphs/graph.py       ← sub-agent graph
    └── graphs/state.py       ← sub-agent state

State Compatibility Check

AgentGraph Compiler automatically compares parent and sub-agent state fields and prints warnings for mismatches:

WARNING: State schema mismatch detected for sub-agent 'rag_agent':
  - Missing on parent state: ['documents', 'web_search']

Add missing fields to parent state_fields to resolve.

11. Generated Output — File by File

`graphs/state.py`

The state schema for this graph. TypedDict by default.

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    question: str
    generation: str

Preservation rule: If all declared fields already exist in the file, the compiler skips regeneration — preserving any manual additions you made.

`graphs/graph.py`

The fully-wired StateGraph. Do not edit manually — regenerate with forge compile.

# Auto-generated LangGraph workflow definition
from langgraph.graph import StateGraph, START, END
from my_agent.graphs.state import AgentState
from my_agent.graphs.nodes.retrieve import retrieve
from my_agent.graphs.nodes.generate import generate

workflow = StateGraph(AgentState)

# Add Nodes
workflow.add_node('retrieve', retrieve)
workflow.add_node('generate', generate)

# Add Edges
workflow.add_edge(START, 'retrieve')
workflow.add_edge('retrieve', 'generate')
workflow.add_edge('generate', END)

# Compile
app = workflow.compile(checkpointer=_checkpointer)

`graphs/nodes/<name>.py`

One file per node. Edit these to implement your business logic.

from my_agent.graphs.state import AgentState
from my_agent.graphs.chains import retrieval_grader

def grade_documents(state: AgentState) -> dict:
    """
    Filters retrieved document chunks and flags irrelevant results.
    """
    print("---RUNNING NODE: grade_documents---")
    # Placeholder: implement node logic here and return a state update dict.
    return {}

`graphs/chains/<name>.py`

One file per chain. Edit to bind your LLM and prompt.

"""
Binary relevance grader.
"""
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# --- LLM Configuration (Anthropic) ---
# from langchain_anthropic import ChatAnthropic
# llm = ChatAnthropic(model="claude-3-5-haiku-20241022", temperature=0.0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Grade the document relevance."),
    ("human", "{input}"),
])

retrieval_grader = prompt | StrOutputParser()

`main.py`

The runnable entrypoint. Streams node events in real time.

import sys
from dotenv import load_dotenv
from my_agent.graphs.graph import app

load_dotenv()

def main():
    question = sys.argv[1] if len(sys.argv) > 1 else "Hello!"
    thread_config = {"configurable": {"thread_id": "main-thread"}}
    inputs = {"question": question}

    print(f"\n🚀 Running agent with: '{question}'\n{'─' * 50}")
    for event in app.stream(inputs, config=thread_config, stream_mode="updates"):
        for node_name, output in event.items():
            print(f"\n📍 [{node_name}]")
            print(output)

if __name__ == "__main__":
    main()

`flowchart.md`

A Mermaid diagram of the compiled graph topology, auto-regenerated on every compile.

12. CLI Reference

`forge compile`

Compile a YAML spec into a Python scaffold.

uv run forge compile configs/agents/agent.yaml           # Full compile
uv run forge compile configs/agents/agent.yaml --dry-run # Preview only (no files written)

--dry-run runs full validation and prints what would be generated without touching the filesystem.

`forge validate`

Validate YAML schema and edge targets without generating files.

uv run forge validate configs/agents/agent.yaml

Catches:

Pydantic schema errors (wrong field types, missing required fields)
Edge targets referencing undefined nodes
supervisor_route edges where route_field is not in state_fields
fan_out edges missing target or field

`forge init`

Scaffold starter config files in the current directory.

uv run forge init
uv run forge init ./my-project

Creates:

configs/config.yaml — global settings template
configs/agents/agent.yaml — starter agent YAML

`forge ingest`

Ingest URLs from configs/config.yaml into a Qdrant vector store.

uv run forge ingest
uv run forge ingest --config configs/config.yaml

13. Global Configuration (`configs/config.yaml`)

Centralized defaults applied across all agents during compilation.

models:
  chat:
    name: gpt-4o          # Default LLM used if node/chain doesn't declare one
    temperature: 0.5
  embedding:
    name: text-embedding-3-small
    provider: openai

vector_store:
  name: qdrant
  url: "http://localhost:6333"
  collection: "my-collection"

ingestion:
  urls:
    - "https://example.com/docs"
    - "https://example.com/blog"

If present, AgentGraph Compiler automatically injects retriever config into compiled agents:

config["retriever"] = {
    "provider": "qdrant",
    "collection": "my-collection",
    "embedding_model": "text-embedding-3-small",
    "top_k": 4
}

14. Complete End-to-End Example

Here is a full Adaptive RAG agent, YAML → generated code → implementation:

YAML (`configs/agents/examples/simple_rag/agent.yaml`)

version: "1.0"

graph:
  source_dir: "simple_rag_agent"
  state_type: typeddict
  state_schema: "graphs.state.RAGState"
  state_fields:
    question: "str"
    generation: "str"
    web_search: "bool"
    documents: "List[str]"
  memory:
    type: in_memory

nodes:
  retrieve:
    module: "graphs.nodes.retrieve"
    function: "retrieve"
    description: "Pulls relevant document chunks from vector store."

  grade_documents:
    module: "graphs.nodes.grade_documents"
    function: "grade_documents"
    chains: ["retrieval_grader"]
    description: "Grades retrieved docs for relevance."

  web_search:
    module: "graphs.nodes.web_search"
    function: "web_search"
    description: "Falls back to Tavily web search."

  generate:
    module: "graphs.nodes.generate"
    function: "generate"
    chains: ["generation_chain"]
    description: "Generates final answer from context."

chains:
  retrieval_grader:
    provider: openai
    llm: gpt-4o-mini
    temperature: 0.0

  generation_chain:
    provider: openai
    llm: gpt-4o
    temperature: 0.2

edges:
  - type: standard
    start: START
    end: retrieve

  - type: standard
    start: retrieve
    end: grade_documents

  - type: conditional
    start: grade_documents
    router:
      module: "routers.router"
      function: "route_after_grading"
    mappings:
      web_search: web_search
      generate: generate

  - type: standard
    start: web_search
    end: generate

  - type: standard
    start: generate
    end: END

Compile

uv run forge compile configs/agents/examples/simple_rag/agent.yaml

Implement Node Logic

# simple_rag_agent/graphs/nodes/retrieve.py
from simple_rag_agent.graphs.state import RAGState
from langchain_qdrant import QdrantVectorStore

retriever = QdrantVectorStore.from_existing_collection(...).as_retriever(k=4)

def retrieve(state: RAGState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"documents": [d.page_content for d in docs]}

# simple_rag_agent/routers/router.py
from simple_rag_agent.graphs.state import RAGState

def route_after_grading(state: RAGState) -> str:
    return "web_search" if state["web_search"] else "generate"

Run

uv run python simple_rag_agent/main.py "What is self-RAG?"

15. Roadmap

v0.3.0 (Planned)

interrupt_after — pause graph after a node completes (mirror of interrupt_before)
Custom state reducers — declare arbitrary reducer functions in YAML (beyond add_messages)
astream() entrypoint — generated main.py uses async def main() + app.astream() when async nodes are present
Stream mode config — stream_mode: values | updates | debug | messages in graph config
Command primitive — nodes can return Command(goto=..., update={}) for dynamic in-node routing
get_state helper — generated utility for inspecting and replaying thread state history
forge migrate — detect state schema drift and guide migration when YAML changes after compile
docker-compose.yaml — agent + Qdrant + Redis sidecar example
MessagesState shorthand — auto-detect single-messages-field state and use built-in MessagesState

Home

AgentGraph Compiler — Developer Wiki

Table of Contents

1. How AgentGraph Compiler Works

Compilation Pipeline

2. YAML Spec — Top-Level Structure

3. graph — Graph Configuration

state_fields Type Mapping

4. State — TypedDict vs Pydantic

TypedDict (Default — LangGraph Native)

Pydantic (Legacy / Advanced Serialization)

5. nodes — Node Types

5.1 Standard Node

5.2 Async Node

5.3 Supervisor Node

5.4 Sub-Agent Reference (graphRef)

5.5 Human Review Node (auto-generated for interrupt_before)

6. chains — LLM Chain Configuration

Supported Providers

7. edges — Edge Types

7.1 standard — Direct Edge

7.2 conditional — Router Function Edge

7.3 supervisor_route — Supervisor Dispatch

7.4 tool_loop — ReAct Tool Loop

7.5 fan_out — Parallel Dispatch (Send API)

7.6 interrupt_before — Human-in-the-Loop

8. tools — Tool Configuration

8.1 Standard LangChain Tool

8.2 MCP Tool Client Stub

9. Memory & Checkpointing

SQLite (Persistent — recommended for production)

In-Memory (Ephemeral — good for development/testing)

Stateless (No Checkpointer)

10. Hierarchical Sub-Agents (graphRef)

How It Works

Output Directory Layout

State Compatibility Check

11. Generated Output — File by File

graphs/state.py

graphs/graph.py

graphs/nodes/<name>.py

graphs/chains/<name>.py

main.py

flowchart.md

12. CLI Reference

forge compile

forge validate

forge init

forge ingest

13. Global Configuration (configs/config.yaml)

14. Complete End-to-End Example

YAML (configs/agents/examples/simple_rag/agent.yaml)

Compile

Implement Node Logic

Run

15. Roadmap

v0.3.0 (Planned)

Clone this wiki locally

3. `graph` — Graph Configuration

`state_fields` Type Mapping

5. `nodes` — Node Types

5.4 Sub-Agent Reference (`graphRef`)

5.5 Human Review Node (auto-generated for `interrupt_before`)

6. `chains` — LLM Chain Configuration

7. `edges` — Edge Types

7.1 `standard` — Direct Edge

7.2 `conditional` — Router Function Edge

7.3 `supervisor_route` — Supervisor Dispatch

7.4 `tool_loop` — ReAct Tool Loop

7.5 `fan_out` — Parallel Dispatch (Send API)

7.6 `interrupt_before` — Human-in-the-Loop

8. `tools` — Tool Configuration

10. Hierarchical Sub-Agents (`graphRef`)

`graphs/state.py`

`graphs/graph.py`

`graphs/nodes/<name>.py`

`graphs/chains/<name>.py`

`main.py`

`flowchart.md`

`forge compile`

`forge validate`

`forge init`

`forge ingest`

13. Global Configuration (`configs/config.yaml`)

YAML (`configs/agents/examples/simple_rag/agent.yaml`)