flowgraph-ai 0.3.8

YAML-driven AI workflow engine powered by LangGraph — build conversational multi-step AI apps without boilerplate

FlowGraph-AI

FlowGraph-AI is a YAML-driven AI workflow engine for building conversational, multi-step AI applications. Define your entire workflow in a simple YAML file — the engine handles LLM calls, user input collection, validation, routing, and state persistence automatically.

Built on top of LangGraph with native interrupt/resume support, SQLite/PostgreSQL persistence, and a built-in FastAPI server.

---

Table of Contents

1. Why FlowGraph-AI?
2. Features
3. Quick Start
4. Installation
5. Core Concepts — 3 Node Types
6. Smart Features
   • Pre-Extraction
   • Tools in DataCollector
   • Follow-Up After Workflow Ends
   • Guardrails
   • Intent Escape
   • Language Detection
   • Hallucination Detection
   • LLM Parameter Customization
7. Kanban Board
8. Task Tracking & Correlation IDs
9. YAML Workflow Reference
10. Routing Reference
11. State Reference
12. Custom Actions
13. Custom Validators
14. CLI Reference
15. REST API Reference
16. Configuration Reference
17. LLM Provider Setup
18. Observability
19. Port Reference
20. Project Structure
21. Testing
22. Changelog
23. License

---

Why FlowGraph-AI?

Most AI frameworks make you write hundreds of lines of code to build a simple conversational flow. FlowGraph-AI flips this — you write a YAML file, the engine does the rest.

Normal chatbot	FlowGraph-AI
Free-form conversation, no structure	YAML-defined flow with clear start → end
Developer writes all conversation logic	Define fields + routing in YAML
Hard to maintain, one giant prompt	Modular nodes, each does one thing
Ask questions one at a time	Extract 3 fields from one natural message
Safety logic must be custom-built	Built-in guardrails on every output
No trace / observability	Langfuse + Jaeger built in
User must re-answer if they said it already	Pre-extraction: bot never re-asks what it knows

workflow: support_ticket
description: "Create a support ticket"

state:
  name: str
  email: str
  issue: str

start_node: collect_info

nodes:
  collect_info:
    type: data_collector
    initial_message: "I'll get that ticket created. What's your name, email, and issue?"
    fields:
      - field: name
        description: "Customer full name"
      - field: email
        description: "Email address"
        validation:
          method: regex
          pattern: "^[\\w.-]+@[\\w.-]+\\.\\w+$"
      - field: issue
        description: "Brief description of the issue"
    max_retry: 3
    next: submit
    on_max_retry: error_response

  submit:
    type: custom_action
    method: "actions.tickets.create"
    routes:
      - value: "created"
        next: success
      - default: error_response

  success:
    type: response
    template: "Ticket created! We'll email {state.email} shortly."

  error_response:
    type: response
    error_field: error

That is the entire application. Run with flowgraph run support_ticket or flowgraph serve.

---

Features

• 3-node architecture — data_collector, custom_action, response — simple enough to reason about, powerful enough for production
• Pre-extraction — if the user's first message already contains a field value, the bot extracts it immediately and never re-asks
• Tools in DataCollector — LLM can call Python tools (DB lookups, APIs, system info) during data collection
• Follow-up handling — after a workflow ends, users can keep asking questions; the bot answers in context
• Hallucination detection — two-stage grounding check with auto-correction, configurable per node
• LLM parameter customization — per-node llm_params: overrides temperature, max_tokens, top_p, streaming, timeout, and model
• args_from_state: for custom actions — clean kwarg-based function calling; maps state fields directly to function parameters
• Real-time Kanban board — React app showing all workflow executions across 8 status columns with live SSE updates
• Task tracking — SQLite-backed task store with pub/sub and correlation IDs linking API, Kanban, Jaeger, and Langfuse
• LangGraph-powered — native interrupt/resume, stateful execution, checkpointer persistence
• SQLite + PostgreSQL — conversations persist across restarts
• Multi-language — auto-detects user language, responds in kind
• Built-in validation — regex, length, numeric, one_of, not_empty + custom Python validators
• Guardrails — LLM safety post-processing on every response, configurable at 3 levels
• 5 LLM providers — Claude, OpenAI, OpenRouter, Ollama, LM Studio; per-node overrides
• FastAPI server — production-ready REST API out of the box
• Developer CLI — flowgraph init, validate, run, serve, test — all from flowgraph
• Observability — Langfuse (LLM tracing) + Jaeger (distributed tracing)

---

Quick Start

Option 1 — Scaffold a new project (recommended)

pip install flowgraph-ai

flowgraph init my-chatbot --provider ollama
cd my-chatbot

ollama pull llama3.1:8b

flowgraph run order_tracking

Try saying: "track my order ORD-001 from yesterday" — both fields extracted at once, no follow-up questions.

Option 2 — From source

git clone https://github.com/simpletoolsindia/flowgraphai
cd flowgraph-ai
uv sync

flowgraph llm set claude       # or ollama, openai, etc.

flowgraph run order_tracking

Option 3 — API server

flowgraph serve
# API at http://localhost:8000 — see /docs for Swagger UI

curl -X POST http://localhost:8000/workflow/start \
  -H "Content-Type: application/json" \
  -d '{"message": "track my order ORD-001 from yesterday"}'

---

Installation

With pip

pip install flowgraph-ai

With uv (recommended)

uv add flowgraph-ai

With UI extras (Streamlit)

pip install "flowgraph-ai[ui]"

From source

git clone https://github.com/simpletoolsindia/flowgraphai
cd flowgraph-ai
uv sync

---

Port Reference

All FlowGraph-AI services use dedicated ports to avoid conflicts. When running the full stack locally, these ports are used:

Port	Service	Description	Start Command
3000	Langfuse	LLM observability UI	flowgraph stack up
3001	Doc-site	Documentation website	cd docs-site && npm run dev
3002	Chat UI	WhatsApp-style testing UI	cd chat-ui && npm run dev
3003	Frontend	Kanban board / task tracker	cd frontend && npm run dev
4317	OTLP gRPC	Jaeger trace collector (gRPC)	flowgraph stack up
4318	OTLP HTTP	Jaeger trace collector (HTTP)	flowgraph stack up
5174	Workflow Builder	Visual workflow editor	cd workflow-builder && npm run dev
5432	PostgreSQL	App database (checkpoints)	External / local install
5433	PostgreSQL	Langfuse database (Docker)	flowgraph stack up
8000	API Server	FlowGraph-AI REST API	flowgraph serve
11434	Ollama	Local LLM server	ollama serve
16686	Jaeger UI	Distributed tracing dashboard	flowgraph stack up

Tip: Port 3000 is reserved for Langfuse. All UI dev servers use different ports to avoid conflicts.

---

Core Concepts — 3 Node Types

FlowGraph-AI has exactly 3 node types. That is intentional.

data_collector — Collect information from the user

Asks questions, waits for replies, validates, extracts, and routes.

Single-field:

collect_sr:
  type: data_collector
  role: "You are a helpful support agent."
  initial_message: "Please provide your SR number (format: SR-XXXXX)."
  field: sr_number
  description: "SR number in format SR-XXXXX"
  max_retry: 3
  next: process
  on_max_retry: error_response
  allow_intent_escape: true
  validation:
    method: "validations.sr.check_sr_format"

Multi-field (collects multiple fields in a single conversation turn):

collect_order:
  type: data_collector
  role: "You are a helpful order support agent."
  initial_message: "Please share your order ID and order date."
  max_retry: 3
  next: lookup_order
  on_max_retry: error_response
  allow_intent_escape: true
  fields:
    - field: order_id
      description: "Order ID (e.g. ORD-12345)"
    - field: order_date
      description: "Order date or approximate date (e.g. yesterday, 10th Dec)"

With tools (LLM calls Python functions for external data):

collect_system_info:
  type: data_collector
  field: cpu_temp
  description: "Current CPU temperature in Celsius"
  tools:
    - method: "actions.system_tools.get_cpu_temp"
      name: "get_cpu_temp"
      description: "Get current CPU temperature. Call when user asks about CPU or system stats."
  next: respond

All data_collector options:

Option	Default	Description
field	—	Single field to collect
fields	—	List of fields (multi-field mode)
description	field name	Tells LLM what to extract
initial_message	LLM-generated	First question shown to user
role	"You are a helpful assistant."	System role for LLM extraction
max_retry	3	Max attempts before routing to on_max_retry
next	"END"	Next node on success
on_max_retry	"END"	Node if max retries exceeded
allow_intent_escape	false	Answer off-topic questions inline before re-asking
humanize	false	Run bot messages through humanizer LLM
validation	—	Validator config (see Custom Validators)
tools	[]	LangChain tools LLM can call during extraction
guardrails	global	Per-node guardrails override
llm_provider	global	Per-node LLM provider override

---

custom_action — Run your Python code

Executes any Python function. The function receives the full state dict, may mutate it, and returns the next node name.

check_account:
  type: custom_action
  method: "actions.billing.check_status"
  result_field: billing_result
  routes:
    - value: "active"
      next: success_response
    - value: "suspended"
      next: suspended_response
    - default: error_response

# actions/billing.py
def check_status(state: dict) -> str:
    account_id = state["account_id"]
    result = billing_api.lookup(account_id)
    state["billing_result"] = result.status
    state["output"] = f"Account {account_id}: {result.status}"
    return result.status   # matched against routes[]

args_from_state — Function keyword arguments (New in v0.2.0)

Clean, kwarg-based function calling. Maps state fields directly to function parameters. The function return value is stored in result_field.

lookup_order:
  type: custom_action
  method: "actions.order_tools.get_order_status"
  args_from_state:
    order_id: order_id          # function_kwarg: state_field
  result_field: order_status    # stores return value
  next: success_response        # fixed routing
  on_error: error_response      # routing on exception

# actions/order_tools.py
def get_order_status(order_id: str) -> str:
    """Explicit args, no state dict needed."""
    status = db.query(order_id)
    return status  # Stored in state["order_status"]

---

response — Send the final reply

The terminal node. Applies guardrails automatically. Supports templates and field references.

success_response:
  type: response
  template: "Done! Your request {state.request_id} has been submitted."

# Or use fields set by custom_action
final:
  type: response
  output_field: output     # state["output"] → shown to user
  error_field: error       # state["error"]  → shown if set

---

Smart Features

Pre-Extraction — No Repeat Questions

Before asking the user anything, the data_collector node checks if the user's opening message already contains the needed field values. If found, extraction happens immediately — no interrupt() is triggered.

Example:

• User sends: "I want to track my order ORD-12345 placed yesterday"
• Node needs: order_id + order_date
• Result: Both fields extracted from the first message — bot never asks either question
• If only order_id is found: bot asks only for order_date

This works for both single-field and multi-field nodes. Validation runs on pre-extracted values; if validation fails, the node falls back to asking normally.

---

Tools in DataCollector

When a field value requires external data (live system info, DB lookups, API calls), define tools: on the node. The LLM decides when to call them based on the user's message.

collect_info:
  type: data_collector
  field: server_status
  description: "Current server status (online/offline/degraded)"
  tools:
    - method: "actions.ops_tools.check_server_status"
      name: "check_server_status"
      description: "Check if a server is online. Provide server_name as argument."
  initial_message: "Which server would you like to check?"
  next: respond

# actions/ops_tools.py
def check_server_status(server_name: str) -> str:
    """Check server health via API."""
    result = ops_api.ping(server_name)
    return result.status  # e.g. "online", "offline", "degraded"

Tool call flow:

1. User says something that needs external data
2. LLM calls the tool (up to 4 rounds of tool calls supported)
3. Tool result is injected into the extraction context
4. LLM uses the tool result to populate the field

Tools also work in off-topic answers — if allow_intent_escape: true and the user's off-topic question needs live data, the same tools are available.

---

Follow-Up After Workflow Ends

After a workflow completes, users can continue asking related questions. The bot answers in the context of the completed conversation.

API behavior: POST /message on a completed session_id returns status: "follow_up" instead of "already completed".

CLI behavior: User can keep typing; the bot answers in context. Only starts a new workflow if the topic is clearly different.

Example:

Bot: Your order ORD-12345 is processing. Delivery expected Dec 15.
     [workflow completes]

User: Why is it taking so long?
Bot:  I understand your concern. Your order was placed yesterday and is currently
      in the processing stage — this typically takes 1-2 business days. Would you
      like me to raise a priority query?

---

Guardrails

Every response node output passes through an LLM safety review before being shown to the user. Configurable at 3 levels (highest priority wins):

node-level guardrails > workflow-level guardrails > global config.yaml

# Global (config.yaml)
guardrails:
  enabled: true
  tone: "warm and professional"

# Workflow-level (top of workflow YAML)
guardrails:
  tone: "casual and friendly"
  custom_rules:
    - "Never mention competitor products"
    - "Always offer to escalate to human support"

# Node-level (inside any response or data_collector node)
my_response:
  type: response
  guardrails:
    enabled: false          # Disable for this node only
    # OR:
    custom_prompt: |        # Fully replaces all guardrail logic
      You are a strict reviewer. Return only the corrected response.

---

Intent Escape

When allow_intent_escape: true, users can ask off-topic questions during data collection. The bot answers inline, then re-asks the original question.

collect_order:
  type: data_collector
  allow_intent_escape: true
  field: order_id
  # ...

Example:

Bot:  Please provide your order ID.
User: Do you ship to Canada?
Bot:  Yes, we ship to Canada! Standard delivery is 5-7 business days.
      Now, could you please provide your order ID?

---

Language Detection

FlowGraph-AI auto-detects the user's language from their first message and injects language instructions into all subsequent LLM prompts. The bot responds in the user's language without any configuration.

Supported: any language the underlying LLM supports (Claude, GPT-4o, Llama 3.1, etc.)

---

Hallucination Detection

Every data collector node can check extracted values for hallucinations before accepting them.

collect_order_info:
  type: data_collector
  hallucination_check: true   # default: true
  fields:
    - field: order_id
      description: "Order ID in format ORD-XXXXX"

The check runs in two stages:

1. Deterministic — substring match verifies the extracted value actually appeared in the user's message.
2. LLM fact-check — if no substring match, an LLM grounding call confirms the value is supported by the conversation context.

When hallucination is detected:

1. Auto-correction using LLM-suggested value (if available)
2. Strict re-extraction with deterministic prompt
3. Retry question to user if both stages fail

Disable per node with hallucination_check: false.

---

LLM Parameter Customization

Customize LLM behaviour globally in config.yaml or per-node in workflow YAML.

Global (config.yaml):

llm:
  provider: "claude"
  temperature: 0.7
  max_tokens: 2048
  top_p: 1.0
  streaming: false
  timeout: 30

Per-node (workflow YAML):

collect_order_info:
  type: data_collector
  llm_params:
    temperature: 0.1      # precise extraction
    max_tokens: 512
    model: "gpt-4o"       # override model for this node only

Any llm_params: key overrides the global config for that node only. All other nodes continue to use global settings.

---

Kanban Board

Track agent execution in real time with the built-in React Kanban board.

Setup

cd frontend
npm install
npm run dev   # Opens on http://localhost:3003

Ensure the backend API is running (flowgraph serve) on port 8000.

Usage

The board connects to the backend API and shows all workflow executions across 8 status columns: Backlog, Planned, In Progress, Waiting For Input, Tool Running, Retry/Recovery, Completed, Failed.

Features:

• Live SSE updates (no polling)
• Click any card to see the full activity timeline
• Correlation ID + Trace ID linking to Jaeger/Langfuse
• Filter by task title, workflow name, or task ID

Every workflow start automatically creates a task card. The card advances through columns as the workflow progresses.

---

Task Tracking & Correlation IDs

Every workflow execution automatically creates a task card backed by SQLite. Tasks are linked across all systems via a correlation_id.

API Response
  └── correlation_id ─────┬──→ Kanban board card
                          ├──→ Jaeger trace root span
                          └──→ Langfuse trace session

API usage

Every POST /workflow/start and POST /message response includes:

{
  "session_id": "order_tracking:abc123",
  "status": "waiting_input",
  "task_id": "A1B2C3D4",
  "correlation_id": "3f8a9c12e4b07d56a1f2e3d4c5b6a7f8"
}

Task lifecycle

Tasks move through these statuses automatically: planned → in_progress → waiting_input ↔ in_progress → completed or failed

When hallucination is detected: retry_recovery → back to in_progress When a tool runs: tool_running → back to in_progress

Configure persistence

# config.yaml
tracking:
  db_path: "./tasks.db"   # SQLite file — omit for in-memory only

---

YAML Workflow Reference

# ── Required ──────────────────────────────────────────────────────────────────
workflow: my_workflow          # Must match the .yaml filename
description: "..."             # Used by intent classifier — describe what this handles

# ── Optional ──────────────────────────────────────────────────────────────────
initial_input_field: user_message   # Which state field gets the user's opening message

# Workflow-level guardrails (override global config.yaml)
guardrails:
  enabled: true
  tone: "casual and friendly"

# Entry point
start_node: first_node

# ── Custom state fields (auto-injected fields do not need to be listed) ───────
state:
  order_id: str
  order_date: str
  order_status: str
  count: int
  items: list
  metadata: dict

# ── Nodes ─────────────────────────────────────────────────────────────────────
nodes:

  # data_collector — single field
  collect_order_id:
    type: data_collector
    role: "You are a helpful support agent."
    initial_message: "Please provide your order ID."
    field: order_id
    description: "Order ID (e.g. ORD-12345)"
    max_retry: 3
    next: process
    on_max_retry: error_response
    allow_intent_escape: true
    humanize: false
    validation:
      method: regex
      pattern: "^ORD-\\d{5}$"
    tools:
      - method: "actions.my_module.my_tool"
        name: "my_tool"
        description: "When and why to call this tool"
    guardrails:
      tone: "very formal"

  # data_collector — multi-field
  collect_contact:
    type: data_collector
    role: "You are an HR assistant."
    initial_message: "Please provide your name and email."
    max_retry: 3
    next: submit
    on_max_retry: error_response
    fields:
      - field: customer_name
        description: "Full name"
      - field: customer_email
        description: "Email address"
        validation:
          method: regex
          pattern: "^[\\w.-]+@[\\w.-]+\\.\\w+$"
      - field: preferred_date
        description: "Preferred contact date"
        default: "any day"    # Optional: used when not provided

  # custom_action
  process:
    type: custom_action
    method: "actions.my_actions.process_request"
    result_field: action_result
    next: success                  # Unconditional (if no routes)
    routes:
      - value: "success"
        next: success_response
      - value: "error"
        next: error_response
      - default: error_response    # Fallback for any other return value
    conditional_edges:             # Alternative to routes
      field: action_result
      paths:
        "ok": success_response
        "fail": error_response

  # response
  success_response:
    type: response
    template: "Done! Your request {state.order_id} has been submitted."
    output_field: output           # Read from state["output"] if set
    error_field: error             # Read from state["error"] if set (shown instead)
    guardrails:
      enabled: false

  error_response:
    type: response
    error_field: error
    output_field: output

---

Routing Reference

Pattern	Where	YAML
Unconditional	data_collector, custom_action	next: node_name
On failure	data_collector	on_max_retry: node_name
Routes list	custom_action	routes: [{value: "x", next: "y"}, {default: "z"}]
Conditional	custom_action	conditional_edges: {field: "f", paths: {v1: n1, v2: n2}}
Direct return	custom_action	Return value IS the next node name

---

State Reference

Auto-injected fields (always available, no need to declare)

Field	Type	Description
user_message	str	User's opening message
user_language	str	Auto-detected language
output	str	Final output shown to user
error	str	Error message (shown instead of output if set)
messages	list	Full conversation history

Template syntax

template: "Your order {state.order_id} placed on {state.order_date} is ready."

All variants work:

• {state.field} — primary
• {{state.field}} — also supported
• {field} — backward compat

---

Custom Actions

# actions/my_actions.py

def process_request(state: dict) -> str:
    """
    Receives full workflow state.
    Mutate state to set output/error values.
    Return a string matched against routes[] in YAML.
    """
    sr = state["sr_number"]
    email = state.get("email", "")

    try:
        result = my_api.submit(sr, email)
        state["output"] = f"Request {sr} submitted. Ref: {result.ref_id}"
        return "success"
    except MyAPIError as e:
        state["error"] = f"Failed: {e.message}"
        return "error"

submit:
  type: custom_action
  method: "actions.my_actions.process_request"
  result_field: action_result
  routes:
    - value: "success"
      next: success_response
    - value: "error"
      next: error_response

---

Custom Validators

# validations/my_validators.py

def validate_account_id(value: str, config: dict) -> tuple[bool, str]:
    """
    Returns (is_valid, error_message).
    error_message is shown to the user on failure.
    """
    if not value.startswith("ACC-"):
        return False, "Account ID must start with ACC- (e.g. ACC-12345)"
    return True, ""

collect_account:
  type: data_collector
  field: account_id
  validation:
    method: "validations.my_validators.validate_account_id"

Built-in validators

Validator	Config	Example
regex	pattern: "..."	pattern: "^\\d{5}$"
length	min: N, max: N	min: 5, max: 200
numeric	min: N, max: N	min: 1, max: 100
one_of	options: [...]	options: [Annual, Sick, Unpaid]
not_empty	—	—

---

CLI Reference

# Project scaffold
flowgraph init [dir] [--provider ollama|claude|openai|...]

# Health & diagnostics
flowgraph health

# Database
flowgraph db set <url>
flowgraph db test

# LLM provider
flowgraph llm set <provider>
flowgraph llm set claude --model claude-opus-4-6
flowgraph llm test

# Workflows
flowgraph list
flowgraph show <name>
flowgraph validate [name]
flowgraph new <name>
flowgraph run <name>
flowgraph run <name> --thread <id>

# API server
flowgraph serve
flowgraph serve --port 9000 --reload

# Observability stack (Docker)
flowgraph stack up
flowgraph stack down
flowgraph stack down --volumes
flowgraph stack status
flowgraph stack logs
flowgraph stack logs langfuse
flowgraph stack configure

# Dev mode
flowgraph dev

# Testing
flowgraph test
flowgraph test --unit
flowgraph test --integration
flowgraph test --coverage
flowgraph test -p tests/unit/test_intent.py
flowgraph test -v

---

REST API Reference

Start with flowgraph serve, then use the Swagger UI at http://localhost:8000/docs.

POST /workflow/start

Start a new conversation. Intent auto-classified if workflow_name is omitted.

curl -X POST http://localhost:8000/workflow/start \
  -H "Content-Type: application/json" \
  -d '{"message": "I need to check my SR status"}'

Response:

{
  "session_id": "order_tracking:abc123",
  "workflow_name": "order_tracking",
  "status": "waiting_input",
  "question": {"field": "order_id", "question": "Please provide your order ID."}
}

Status values: waiting_input | completed | error

POST /message

Resume a session or send a follow-up after completion. When the session is already completed, returns status: "follow_up" with a contextual answer.

GET /session/{session_id}

Inspect current state of a session.

GET /workflows

List all workflows with descriptions.

GET /health

System health + configuration summary.

GET /workflow/{name}/mermaid

Mermaid diagram for a workflow.

GET /tasks

List all tasks. Supports optional status query param filter.

GET /tasks/{task_id}

Get a single task by ID including its full event timeline.

GET /tasks/session/{session_id}

Get all tasks associated with a session ID.

GET /tasks/stream/events

Server-Sent Events stream used by the Kanban board for live updates.

All WorkflowResponse objects include task_id and correlation_id fields.

---

Configuration Reference

config.yaml — controls all runtime behaviour:

app:
  name: FlowGraph-AI
  version: 0.1.0

# ── LLM Provider ──────────────────────────────────────────────────────────────
llm:
  provider: ollama            # claude | openai | openrouter | lmstudio | ollama
  temperature: 0.7
  max_tokens: 2048            # Global default (per-node llm_params: overrides)
  top_p: 1.0
  streaming: false
  timeout: 30                 # Seconds

  ollama_model: llama3.1:8b
  ollama_base_url: http://localhost:11434

  claude_model: claude-sonnet-4-6
  # ANTHROPIC_API_KEY env var

  openai_model: gpt-4o-mini
  # OPENAI_API_KEY env var

  openrouter_model: openai/gpt-4o-mini
  openrouter_base_url: https://openrouter.ai/api/v1
  # OPENROUTER_API_KEY env var

  lmstudio_model: local-model
  lmstudio_base_url: http://localhost:1234/v1

# ── Engine ────────────────────────────────────────────────────────────────────
engine:
  max_steps: 50               # Max LangGraph execution steps per invocation

# ── Intent Classifier ─────────────────────────────────────────────────────────
intent:
  fallback_workflow: greeting   # Workflow used when no intent matches

# ── Guardrails ────────────────────────────────────────────────────────────────
guardrails:
  enabled: true
  tone: "warm and professional"
  # custom_rules:
  #   - "Never mention competitor products"
  # custom_prompt: |           # Fully replaces tone + custom_rules
  #   You are a strict reviewer...

# ── Language Detection ────────────────────────────────────────────────────────
language_detection:
  enabled: true               # Detect language + respond in kind

# ── Logging ───────────────────────────────────────────────────────────────────
logging:
  level: WARNING              # DEBUG | INFO | WARNING | ERROR
  format: text                # text | json

# ── Checkpointing ────────────────────────────────────────────────────────────
checkpoint:
  connection_string: ""       # Empty = in-memory (dev)
  # connection_string: "./flowgraph.db"          # SQLite
  # connection_string: "postgresql://user:pass@localhost:5432/mydb"  # PostgreSQL

# ── Task Tracking ─────────────────────────────────────────────────────────────
tracking:
  db_path: "./tasks.db"       # SQLite file for task history; omit = in-memory

# ── Observability ─────────────────────────────────────────────────────────────
observability:
  langfuse:
    enabled: false
    host: http://localhost:3000
    public_key: ""            # or env: LANGFUSE_PUBLIC_KEY
    secret_key: ""            # or env: LANGFUSE_SECRET_KEY
  tracing:
    enabled: false
    endpoint: http://localhost:4317   # Jaeger OTLP gRPC
    service_name: flowgraph-ai

Environment variables

Variable	Purpose
ANTHROPIC_API_KEY	Claude API key
OPENAI_API_KEY	OpenAI API key
OPENROUTER_API_KEY	OpenRouter API key
LANGFUSE_PUBLIC_KEY	Langfuse public key
LANGFUSE_SECRET_KEY	Langfuse secret key
FLOWGRAPH_LOG_LEVEL	Override log level

Set these in a .env file in the project root — auto-loaded at startup.

Per-node LLM override

my_node:
  type: data_collector
  llm_provider: claude        # Override global provider for this node only
  field: name

---

LLM Provider Setup

Ollama (local, free — default)

brew install ollama            # macOS
curl -fsSL https://ollama.ai/install.sh | sh   # Linux

ollama pull llama3.1:8b
ollama serve

llm:
  provider: ollama
  ollama_model: llama3.1:8b

Recommended models:

• llama3.1:8b — fast, works for most flows
• llama3.1:70b — much better JSON extraction (needed for multi-field on small models)
• mistral:7b — good structured output

Claude (best quality)

echo "ANTHROPIC_API_KEY=sk-ant-..." >> .env

llm:
  provider: claude
  claude_model: claude-sonnet-4-6

OpenAI

echo "OPENAI_API_KEY=sk-..." >> .env

llm:
  provider: openai
  openai_model: gpt-4o-mini

OpenRouter (access any model via one API)

echo "OPENROUTER_API_KEY=sk-or-..." >> .env

llm:
  provider: openrouter
  openrouter_model: anthropic/claude-3.5-sonnet
  openrouter_base_url: https://openrouter.ai/api/v1

LM Studio (local)

Start LM Studio, load a model, then:

llm:
  provider: lmstudio
  lmstudio_model: local-model
  lmstudio_base_url: http://localhost:1234/v1

---

Observability

Services

Service	URL	Purpose
Langfuse UI	http://localhost:3000	LLM prompts, completions, tokens, cost, latency
Jaeger UI	http://localhost:16686	Distributed traces: API → workflow → node spans
OTLP gRPC	localhost:4317	App sends OTel spans here
PostgreSQL	localhost:5433	Langfuse storage (separate from app DB)

What gets tracked

Langfuse — every LLM call:

• Full prompt + completion
• Token usage + cost estimate
• Latency per call
• Session grouping by thread_id
• Trace name: workflow:<name>

Jaeger — every workflow execution:

• Root span per API request
• Node-level child spans
• Workflow status + session ID attributes

---

Project Structure

As a developer, you only work in:

• workflows/ — define conversations in YAML
• actions/ — write Python business logic
• validations/ — write custom validators
• config.yaml — configure LLM, guardrails, logging

---

Testing

flowgraph test
flowgraph test --coverage
flowgraph test --integration
flowgraph test -p tests/unit/test_intent.py
flowgraph test -v

Tests live in:

• tests/unit/ — fast, isolated, no API keys needed (754 tests, 86% coverage)
• tests/integration/ — test full graph execution with real LLM + DB

---

Changelog

See CHANGELOG.md for the full history.

[0.2.1] — 2026-03-12

• Real-time Kanban board (frontend/) — 8-column board with SSE live updates and activity timeline
• Task tracking (flowgraph_ai/tracking/) — SQLite pub/sub task store; auto-creates a card on every workflow start
• Hallucination detection (engine/hallucination.py) — two-stage grounding check with auto-correction
• LLM parameter customization — per-node llm_params: for temperature, max_tokens, top_p, model, etc.
• args_from_state for custom_action — clean kwarg-based function calling; maps state fields to function params
• Correlation IDs — links API response, Kanban card, Jaeger trace, and Langfuse trace
• 4 new task API endpoints — GET /tasks, GET /tasks/{id}, GET /tasks/session/{id}, GET /tasks/stream/events
• 754 unit tests passing; 86% code coverage
• Backward compatible — no breaking changes from v0.2.0

[0.2.0] — 2026-03-12

• Pre-extraction — extracts field values from the user's first message before asking
• Tools in DataCollector — LLM calls Python tools during data collection (multi-round loop)
• Follow-up handling — POST /message on a completed session returns status: "follow_up"
• flowgraph init command — scaffolds a complete project with two ready-to-run examples
• GitHub Actions CI/CD — unit tests, lint, package build on push/PR
• 155 unit tests passing

---

License

MIT — see LICENSE

Bug reports and contributions welcome. Include flowgraph health output + full error message with bug reports.

• Email: [e-mail]
• Issues: https://github.com/simpletoolsindia/flowgraphai/issues