Skip to content

Home

Jump to bottom
Chay edited this page Jul 24, 2026 · 4 revisions

GraphForge — Developer Wiki

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

Table of Contents

  1. How GraphForge Works
  2. YAML Spec — Top-Level Structure
  3. graph — Graph Configuration
  4. State — TypedDict vs Pydantic
  5. nodes — Node Types
  6. chains — LLM Chain Configuration
  7. edges — Edge Types
  8. tools — Tool Configuration
  9. Memory & Checkpointing
  10. Hierarchical Sub-Agents (graphRef)
  11. Generated Output — File by File
  12. CLI Reference
  13. Global Configuration (configs/config.yaml)
  14. Complete End-to-End Example
  15. Roadmap
  16. How AgentGraph Compiler Works
  17. CLI Usage & Workflows
  18. YAML Specification Reference
  19. Architecture Patterns & Scaffolding
  20. State Schemas & Reducers
  21. Sub-Agent & Parallel Fan-Out Deep Dive
  22. Vector Store & Retrieval Architecture
  23. Generated Code Anatomy

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

  1. Parse YAML → Python dict
  2. Validate schema (Pydantic) → check edge targets, supervisor route fields, fan_out params
  3. Create directories → graphs/, nodes/, chains/, tools/, routers/
  4. Generate graphs/state.py → TypedDict or Pydantic state class
  5. Scaffold nodes → one .py file per node with typed function stub
  6. Scaffold chains → one .py file per chain with LLM binding comments
  7. Scaffold tools → one .py file per tool (or MCP client stub)
  8. Scaffold routers → conditional edge router function stubs
  9. Generate graphs/graph.py → fully-wired StateGraph with all edges, ToolNode, Send fan-outs, checkpointer
  10. Generate main.py → runnable entrypoint with app.stream()
  11. 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. GraphForge 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, GraphForge 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). GraphForge 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: GraphForge 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:

  1. rag_agent.yaml is compiled first, generating code in my_agent/rag_agent/
  2. 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

GraphForge 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, GraphForge 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

Clone this wiki locally