Metadata-Version: 2.4
Name: flowgraph-ai
Version: 0.1.0
Summary: YAML-driven AI workflow engine powered by LangGraph — build conversational multi-step AI apps without boilerplate
Author-email: Simpletools India <support@simpletools.in>
License-Expression: MIT
Project-URL: Homepage, https://github.com/simpletoolsindia/flowgraphai
Project-URL: Repository, https://github.com/simpletoolsindia/flowgraphai
Project-URL: Documentation, https://github.com/simpletoolsindia/flowgraphai#readme
Project-URL: Changelog, https://github.com/simpletoolsindia/flowgraphai/blob/main/CHANGELOG.md
Project-URL: Bug Tracker, https://github.com/simpletoolsindia/flowgraphai/issues
Keywords: ai,llm,workflow,langgraph,langchain,chatbot,conversational-ai,yaml,fastapi,openai,anthropic,ollama,agent,automation
Classifier: Development Status :: 2 - Pre-Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Framework :: FastAPI
Classifier: Typing :: Typed
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: langchain-anthropic>=0.1.0
Requires-Dist: langchain-ollama>=0.2.0
Requires-Dist: langchain-openai>=0.1.0
Requires-Dist: langgraph>=0.2.0
Requires-Dist: langgraph-checkpoint-sqlite>=2.0.0
Requires-Dist: langgraph-checkpoint-postgres>=2.0.0
Requires-Dist: fastapi>=0.115.0
Requires-Dist: uvicorn>=0.32.0
Requires-Dist: python-dotenv>=1.0.0
Requires-Dist: pyyaml>=6.0.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: langfuse>=2.0.0
Requires-Dist: opentelemetry-sdk>=1.28.0
Requires-Dist: opentelemetry-exporter-otlp-proto-grpc>=1.28.0
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.28.0
Provides-Extra: postgres
Requires-Dist: psycopg[binary]>=3.2.0; extra == "postgres"
Requires-Dist: psycopg-pool>=3.2.0; extra == "postgres"
Requires-Dist: langgraph-checkpoint-postgres>=2.0.0; extra == "postgres"
Provides-Extra: ui
Requires-Dist: streamlit>=1.55.0; extra == "ui"
Requires-Dist: streamlit-mermaid>=0.3.0; extra == "ui"
Provides-Extra: all
Requires-Dist: flowgraph-ai[ui]; extra == "all"
Requires-Dist: flowgraph-ai[postgres]; extra == "all"
Dynamic: license-file

# FlowGraph-AI

[![PyPI version](https://img.shields.io/pypi/v/flowgraph-ai.svg)](https://pypi.org/project/flowgraph-ai/)
[![Python](https://img.shields.io/pypi/pyversions/flowgraph-ai.svg)](https://pypi.org/project/flowgraph-ai/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![LangGraph](https://img.shields.io/badge/powered%20by-LangGraph-blue)](https://github.com/langchain-ai/langgraph)

**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.

---

## 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.

```yaml
workflow: support_ticket
description: "Create a support ticket"

state:
  name: str
  email: str
  issue: str

start_node: collect_name

nodes:
  collect_name:
    type: data_collector
    initial_message: "What is your name?"
    field: name
    next: collect_email
    on_max_retry: error_response

  collect_email:
    type: data_collector
    initial_message: "What is your email address?"
    field: email
    validation:
      method: regex
      pattern: "^[\\w.-]+@[\\w.-]+\\.\\w+$"
    next: submit
    on_max_retry: error_response

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

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

  error_response:
    type: response
    error_field: error
```

That is the entire application. Run it with `flowgraph run support_ticket` or expose it via the built-in API server.

---

## Features

- **3-node architecture** — `data_collector`, `custom_action`, `response` — simple enough to reason about, powerful enough for production
- **YAML-first** — define complete multi-step AI workflows without boilerplate
- **Python builder API** — fluent code alternative to YAML for programmatic workflow construction
- **LangGraph-powered** — native interrupt/resume, stateful execution, checkpointer persistence
- **SQLite + PostgreSQL** — conversations persist across restarts, zero extra config for SQLite
- **Multi-language** — auto-detects user language, responds in kind across all LLM calls
- **Intent classification** — automatically routes users to the right workflow based on what they say
- **Built-in validation** — regex, length, numeric, one_of, not_empty — plus custom Python validators
- **Guardrails** — LLM safety post-processing on every response, configurable
- **5 LLM providers** — Claude, OpenAI, OpenRouter, Ollama, LM Studio; per-node overrides
- **FastAPI server** — production-ready REST API out of the box
- **Developer CLI** — scaffold, validate, run, serve — all from `flowgraph`
- **Observability** — Langfuse (LLM tracing) + Jaeger (distributed tracing) with one command setup

---

## Installation

### With pip

```bash
pip install flowgraph-ai
```

### With uv (recommended)

```bash
uv add flowgraph-ai
```

### From source

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

---

## Quick Start

### 1. Set your LLM provider

```bash
# Claude (Anthropic)
export ANTHROPIC_API_KEY=sk-ant-...
flowgraph llm set claude

# OpenAI
export OPENAI_API_KEY=sk-...
flowgraph llm set openai

# Local (Ollama, no key needed)
flowgraph llm set ollama
```

### 2. Create a workflow

```bash
flowgraph new my_workflow
```

Or write your own `workflows/my_workflow.yaml`.

### 3. Run it

```bash
# Interactive CLI
flowgraph run my_workflow

# API server
flowgraph serve

# Full dev mode (API + observability stack)
flowgraph dev
```

---

## Core Concepts

### Node Types

FlowGraph-AI has exactly **3 node types**. That is intentional — simplicity over abstraction.

#### `data_collector` — Collect information from the user

Asks a question, waits for the user's reply, validates it, and routes to the next node.

```yaml
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    # Answer off-topic questions before re-asking
  validation:
    method: "validations.sr.check_sr_format"
```

**Multi-field collection** (collects multiple fields in one prompt):

```yaml
collect_dates:
  type: data_collector
  role: "You are an HR assistant."
  initial_message: "Please provide your leave start and end dates."
  fields:
    - field: start_date
      description: "Leave start date (YYYY-MM-DD)"
      validation:
        method: regex
        pattern: "^\\d{4}-\\d{2}-\\d{2}$"
    - field: end_date
      description: "Leave end date (YYYY-MM-DD)"
      validation:
        method: regex
        pattern: "^\\d{4}-\\d{2}-\\d{2}$"
  max_retry: 3
  next: submit
  on_max_retry: error_response
```

#### `custom_action` — Run your Python code

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

```yaml
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
```

The Python function:

```python
# 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[]
```

#### `response` — Send the final reply

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

```yaml
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
```

---

### Routing

| Pattern | YAML | When to use |
|---|---|---|
| Unconditional | `next: node_name` | Always go to one node |
| On-failure | `on_max_retry: node_name` | data_collector max retries exceeded |
| Routes list | `routes: [{value: "x", next: "y"}, {default: "z"}]` | Match custom_action return value |
| Conditional | `conditional_edges: {field: "result", paths: {...}}` | Route on field value |

---

### State

Every workflow has auto-injected standard state fields. Add your own custom fields in the `state:` block.

```yaml
state:
  # Your custom fields
  account_id: str
  invoice_total: float
  items: list

# Auto-injected (always available):
#   user_message   — the user's first message
#   user_language  — detected language (e.g. "English", "Spanish")
#   output         — final response to show user
#   error          — error message (shown instead of output if set)
#   messages       — full conversation history list
```

Access state in templates: `{state.account_id}`

---

## YAML Workflow Reference

```yaml
workflow: my_workflow          # Internal name (must match filename)
description: "..."             # Used by intent classifier to route users here

# Initial field to populate with the user's first message (default: user_message)
initial_input_field: user_message

# Custom state fields
state:
  my_field: str
  count: int
  items: list

# Entry point
start_node: first_node

nodes:
  first_node:
    type: data_collector | custom_action | response
    # ... node-specific config
```

---

## Python Builder API

For programmatic workflow construction without YAML:

```python
from engine.workflow_builder import workflow
from llm.provider import get_llm
from config import load_config
from storage.checkpointer import get_checkpointer

cfg = load_config()
llm = get_llm(cfg)
checkpointer = get_checkpointer(cfg)

graph = (
    workflow()
    .state(account_id="str", result="str")
    .collect(
        name="collect_id",
        field="account_id",
        ask="Please provide your account ID.",
        next_node="process",
        on_max_retry="error",
    )
    .action(
        name="process",
        method="actions.billing.check_status",
        result_field="result",
        routes=[{"value": "ok", "next": "done"}, {"default": "error"}],
    )
    .respond(name="done",  template="All good! Account: {state.account_id}")
    .respond(name="error", error_field="error")
    .start("collect_id")
    .compile(llm=llm, app_config=cfg, checkpointer=checkpointer)
)
```

---

## Custom Actions

```python
# actions/my_actions.py

def process_request(state: dict) -> str:
    """
    Receives full workflow state.
    Mutate state to set output 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 to submit: {e.message}"
        return "error"
```

```yaml
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

```python
# 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 user on failure.
    """
    if not value.startswith("ACC-"):
        return False, "Account ID must start with ACC- (e.g. ACC-12345)"
    if len(value) != 9:
        return False, "Account ID must be exactly 9 characters (e.g. ACC-12345)"
    return True, ""
```

```yaml
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

```bash
# Health & diagnostics
flowgraph health               # Full system health check (DB, LLM, workflows, observability)

# Database
flowgraph db set <url>         # Set checkpoint DB (postgresql://... or ./file.db or empty for memory)
flowgraph db test              # Test database connection

# LLM provider
flowgraph llm set <provider>   # Switch provider: claude | openai | openrouter | lmstudio | ollama
flowgraph llm set claude --model claude-opus-4-6
flowgraph llm test             # Send test prompt to LLM

# Workflows
flowgraph list                 # List all workflows with descriptions
flowgraph show <name>          # Show workflow node graph + state schema
flowgraph validate [name]      # Validate YAML syntax and routing
flowgraph new <name>           # Scaffold a new workflow interactively
flowgraph run <name>           # Chat with a workflow in the terminal
flowgraph run <name> --thread <id>  # Resume an existing conversation

# API server
flowgraph serve                # Start the FastAPI server (default: 0.0.0.0:8000)
flowgraph serve --port 9000 --reload

# Observability stack (Docker)
flowgraph stack up             # Start PostgreSQL + Langfuse + Jaeger
flowgraph stack down           # Stop the stack
flowgraph stack status         # Show container health + service URLs
flowgraph stack logs           # Tail all service logs
flowgraph stack logs langfuse  # Tail specific service logs
flowgraph stack configure      # Save Langfuse API keys + enable observability

# One-command full dev start
flowgraph dev                  # stack up + API server + show all URLs
```

---

## REST API

Start with `flowgraph serve` then hit:

### POST `/workflow/start`
Start a new conversation. Intent is auto-classified if `workflow_name` is omitted.

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

```json
{
  "session_id": "service_request:abc123",
  "workflow_name": "service_request",
  "status": "waiting_input",
  "question": {"field": "sr_number", "question": "Please provide your SR number (SR-XXXXX)."}
}
```

### POST `/message`
Resume a session with the user's reply.

```bash
curl -X POST http://localhost:8000/message \
  -H "Content-Type: application/json" \
  -d '{"session_id": "service_request:abc123", "message": "SR-12345"}'
```

### GET `/session/{session_id}`
Inspect current state of a session.

### GET `/workflows`
List all available workflows.

### GET `/health`
System health + configuration summary.

### GET `/workflow/{name}/mermaid`
Get Mermaid diagram for a workflow (for visualization).

---

## Configuration

`config.yaml` controls all runtime behaviour:

```yaml
app:
  name: FlowGraph-AI
  version: 0.1.0

llm:
  provider: claude              # claude | openai | openrouter | lmstudio | ollama
  temperature: 0.7
  claude_model: claude-sonnet-4-6
  openai_model: gpt-4o-mini
  ollama_model: llama3.1:8b
  ollama_base_url: http://localhost:11434

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

intent:
  fallback_workflow: greeting   # Workflow used when no intent matches

guardrails:
  enabled: true                 # LLM safety review on every response node output

language_detection:
  enabled: true                 # Auto-detect user language; respond in kind

checkpoint:
  connection_string: ./flowgraph.db       # SQLite (dev)
  # connection_string: postgresql://user:pass@localhost:5432/mydb  # PostgreSQL (prod)

observability:
  langfuse:
    enabled: false              # Enable after: flowgraph stack configure
    host: http://localhost:3000
    public_key: ""              # or env: LANGFUSE_PUBLIC_KEY
    secret_key: ""              # or env: LANGFUSE_SECRET_KEY
  tracing:
    enabled: false              # Enable after: flowgraph stack up
    endpoint: http://localhost:4317
    service_name: flowgraph-ai
```

### Per-node LLM override

Any node can use a different LLM provider:

```yaml
collect_name:
  type: data_collector
  llm_provider: claude          # Override global provider for this node only
  field: name
  ...
```

---

## Observability

FlowGraph-AI includes first-class observability with zero manual Docker setup.

### Langfuse — LLM Tracing

Tracks every LLM call: full prompts, completions, token usage, cost, latency.

```bash
flowgraph stack up             # Starts Langfuse (+ PostgreSQL + Jaeger) via Docker
# Open http://localhost:3000 → create account → copy API keys
flowgraph stack configure      # Saves keys to config.yaml, enables tracing
flowgraph serve                # API now sends all LLM traces to Langfuse
```

### Jaeger — Distributed Tracing

Traces every API request → workflow execution → node step as nested spans.

Starts automatically with `flowgraph stack up`. View traces at **http://localhost:16686**.

### What gets tracked

| System | Captures |
|---|---|
| **Langfuse** | LLM prompts, completions, tokens, cost, latency per call, session grouping |
| **Jaeger** | API request spans, workflow spans, node spans, error recording |

---

## Adding a New Workflow

1. Create `workflows/my_workflow.yaml`
2. Add custom actions in `actions/my_actions.py` (if needed)
3. Add custom validators in `validations/my_validators.py` (if needed)
4. Test: `flowgraph validate my_workflow` then `flowgraph run my_workflow`

The intent classifier discovers new workflows automatically — no code changes needed.

---

## Architecture

```
User / Client
    ↓
FastAPI  (api/server.py)
    ↓
Intent Classifier  (intent.py)  — routes message to correct workflow
    ↓
Graph Builder  (engine/graph_builder.py)  — builds LangGraph StateGraph from YAML
    ↓
LangGraph StateGraph
    ├── data_collector   — interrupt() / resume, field extraction, validation
    ├── custom_action    — executes Python method, routes on return value
    └── response         — renders output, applies LLM guardrails
    ↓
Checkpointer  (storage/checkpointer.py)  — SQLite / PostgreSQL / memory
    ↓
LLM Provider  (llm/provider.py)  — Claude / OpenAI / Ollama / ...
    ↓
Observability
    ├── Langfuse  — LLM call tracing (callbacks injected at invocation)
    └── Jaeger    — Distributed request tracing (OpenTelemetry spans)
```

---

## Supported LLM Providers

| Provider | Key | How |
|---|---|---|
| **Anthropic Claude** | `claude` | `ANTHROPIC_API_KEY` |
| **OpenAI** | `openai` | `OPENAI_API_KEY` |
| **OpenRouter** | `openrouter` | `OPENROUTER_API_KEY` |
| **LM Studio** | `lmstudio` | No key (local) |
| **Ollama** | `ollama` | No key (local) |

---

## License

MIT — see [LICENSE](LICENSE)

---

## Contributing

Contributions are welcome. Please open an issue before submitting a PR.

Bug reports: include `flowgraph health` output + the full error message.
Email: support@simpletools.in
Issues: https://github.com/simpletoolsindia/flowgraphai/issues
