Flowfile AI Subsystem Guide

Flowfile AI subsystem

Everything under flowfile_core/flowfile_core/ai/ — the /ai/* HTTP surface, the three agent tiers (assist / copilot / planner), the litellm provider seam, BYOK credentials, rate limiting, the prompt log, and the local model manager. Read this before touching any file under flowfile_core/flowfile_core/ai/ or flowfile_core/tests/ai/.

When NOT to use this skill

• The full FEATURE_FLAG_AI / FLOWFILE_AI_LOG_PROMPTS* / rate-limit env-var table with exact settings.py line numbers → flowfile-config-and-flags (this skill states only what's needed to reason about AI behavior).
• Step-by-step "an agent is misbehaving, what do I check first" triage → flowfile-debugging-playbook (this skill explains why the prompt log works the way it does; the playbook is the symptom-first runbook).
• Adding a brand-new node type (Pydantic settings, add_<type> on FlowGraph, frontend node UI) → flowfile-node-development. The AI tool catalog auto-derives from those same settings classes — get the node right first, the AI surfaces it for free.
• Core/worker/kernel dataflow contract (why core never .collect()s) → flowfile-architecture-contract.
• Frontend chat/copilot UI components (AiAssistant.vue, ghost-node suggestions, Cmd+K palette rendering) → flowfile-frontend-conventions.
• Test markers and how to run flowfile_core/tests/ai/ → flowfile-testing-and-validation.

---

1. Surface map — three agent tiers, one prompt stack

All AI code lives under flowfile_core/flowfile_core/ai/. The package docstring (ai/__init__.py) is the map: providers/ (litellm seam + BYOK key load), tools/ (tool catalog + executor), context/ (subgraph/schema serialization), agents/ (surface implementations), prompts/ (layered system prompts), plus streaming, scheduler, sessions, diff, safety, metrics. 17 *_routes.py route modules (as of 2026-07-03; re-count with ls flowfile_core/flowfile_core/ai/*_routes.py) are mounted onto one router, which is the package's only public export.

Do not look for assist/copilot implementation inside agents/assist.py or agents/copilot.py — they only point you at the real route modules and at prompts/base.md + prompts/assist.md / prompts/copilot.md. All three tiers share prompts/base.md as a common layer.

Planner surfaces (verify before quoting elsewhere — this drifted once already)

AgentStartRequest.surface (agent_routes.py) is a Literal["agent_complex", "agent_staged", "agent_live"], default "agent_staged" — not a bare "agent". agents/planner/_internal.py defines _STAGED_STATE_MACHINE_SURFACES = frozenset({"agent_staged", "agent_live"}):

• agent_staged (default) — the two-stage state machine: classify → pick_type → pick_upstream → fill_settings. Every tool call dispatches with mode="stage"; nothing touches the live graph until the user accepts.
• agent_live — same state machine, but applies each step immediately (lives in flow.nodes right away) instead of staging, and includes an observation step. Every agent_staged gate in the planner code must also include agent_live (see the comment at _internal.py:65-69).
• agent_complex — single-shot, full tool catalog (including codegen helpers hidden from the staged surfaces). Requires a provider with supports_tools=True, or the route returns 422.

Planner contract, verified against agents/planner/__init__.py's module docstring:

• Session opens via POST /ai/agent/start; every LLM tool call dispatches through execute_tool_call with mode="stage" — the live graph is never mutated mid-run (staged surfaces) or mutated one step at a time with full auditing (agent_live).
• The graph is snapshotted before every dispatch. If the user edits the canvas mid-run, the planner yields drift_detected + paused and exits cleanly; the session waits for POST /ai/agent/{session_id}/resume ({"action": "continue"} re-snapshots and resumes; {"action": "discard"} frees the staged diff).
• A rejected step retries up to max_retries_per_step times by feeding the executor's refusal_detail back to the LLM as a role="tool" message.
• On completion, per-step results bundle into one flowfile_core.ai.diff.GraphDiff via bundle_staged_results, reviewed and accepted atomically in the UI (AiDiffPreview).
• The planner generator never raises — every failure becomes a PlannerEvent ("error" / "tool_call_rejected" / "drift_detected" / "abort"). SSE id: headers are f"{session_id}.{step_count}" so Last-Event-ID reconnect works against the replay buffer.
• Sessions are namespaced by user_id; cross-user access on {session_id}/* returns 404 (not 403) to avoid leaking session existence.

---

2. Router gating — FEATURE_FLAG_AI

The entire /ai/* router is behind one dependency: ai/feature_flag.py::require_ai_enabled(), wired as APIRouter(dependencies=[Depends(require_ai_enabled)]) in ai/routes.py. Off → every AI endpoint returns 503 with detail "AI features are disabled. Set FEATURE_FLAG_AI=true to enable." (DISABLED_DETAIL, feature_flag.py:36).

• Backing store: settings.FEATURE_FLAG_AI, a MutableBool (configs/settings.py:25-27), default on (os.environ.get("FEATURE_FLAG_AI", "1"), truthy strings true/1/yes/on, case/whitespace-insensitive).
• is_ai_enabled() reads the live MutableBool value on every call — no restart, no reload needed to see a flip.
• Live flip: POST /system/feature_flags/ai (ai/admin_routes.py, admin-only via get_current_admin_user) — mounted under /system, deliberately outside the gated /ai router (so you can turn AI back on while it's reporting 503). It sets both settings.FEATURE_FLAG_AI.set(enabled) and os.environ["FEATURE_FLAG_AI"], but the response always carries persisted: false — surviving a process restart is the operator's .env file, not this endpoint's job.
• Full env-var table (defaults, truthy-parse rule, every other FLOWFILE_AI_* flag) → flowfile-config-and-flags.

---

3. The litellm lazy-import contract

Rule: no module under ai/ imports litellm at module (top) level, so the package can be imported — including by cheap classification/dry-run/ scheduling code paths that never need a real LLM call — without paying litellm's import cost or triggering its network calls.

Verified (grep for import litellm / from litellm import under flowfile_core/flowfile_core/ai/) — there are exactly two real imports, both function-local:

1. ai/providers/_litellm_base.py, inside _lazy_litellm() — import litellm then litellm.suppress_debug_info = True (idempotent; silences litellm's stdout banners on every call).
2. ai/scheduler.py, inside a function — from litellm import exceptions as lle # lazy, guarded by try/except so a missing/broken litellm install degrades to bare TimeoutError/ConnectionError retry matching instead of crashing.

ai/__init__.py also does os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True") before importing routes — a write, not an import — guaranteeing litellm's eventual first import uses the bundled cost map instead of fetching one from raw.githubusercontent.com.

Correction to a common paraphrase of the root CLAUDE.md line ("keep the package litellm-import-free except ai/byok.py"): read literally that's wrong — byok.py contains no litellm import at all (verified). What byok.py's own docstring actually describes is that it keeps ai/credentials.py (the DB CRUD layer) provider-class-import-free, so import flowfile_core.ai.credentials never pulls in any providers/* module. The real litellm rule is simpler and package-wide: no module under ai/ imports litellm at load time; the two lazy sites above are it.

Enforcing tests (flowfile_core/tests/ai/) — re-adding an eager import litellm anywhere under ai/ breaks these:

• test_classification.py::test_classify_lazy_litellm — asserts no litellm/litellm.* module is in sys.modules after importing the classification module.
• test_dry_run.py::test_dry_run_lazy_litellm_contract — same assertion for the dry-run module.
• ai/scheduler.py's own docstring states the same invariant for flowfile_core.ai.scheduler, but no dedicated test for it was found beyond the two above — don't assume a third exists.

LiteLLMProvider.chat() / .stream() (ai/providers/_litellm_base.py) are the single dispatch seam every vendor subclass shares: translate Pydantic Message/ToolSpec to litellm's dict shapes, aggregate streamed tool-call fragments in _PartialToolCall buffers (a tool call surfaces only once it has an id + name and its arguments parse as JSON), and in finally blocks run two never-crash hooks — _record_call_telemetry (always-on flowfile_ai_provider_call_total counter) and _record_chat_log/ _record_stream_log (prompt log, only when FLOWFILE_AI_LOG_PROMPTS is on). Both import their dependencies function-locally and swallow + log failures — a logging bug must never take down an LLM call.

---

4. Providers, BYOK, and model resolution

ai/providers/registry.py::PROVIDERS maps six litellm-backed vendors — anthropic, openai, google, groq, openrouter, ollama — each a config-only subclass of LiteLLMProvider (name, default_model, model_prefix, capability flags, surface_models, default_api_base).

The local llama.cpp pseudo-provider (providers/local.py, LOCAL_PROVIDER_ID = "local") is deliberately not in PROVIDERS — it never appears in BYOK credential CRUD. is_resolvable_provider(name) = PROVIDERS ∪ {"local"}, used by read-only routes; the tool-calling agent route (agent_routes.py) rejects provider == LOCAL_PROVIDER_ID explicitly — a local model has no reliable tool-calling support to build a planner run on.

BYOK (ai/byok.py + ai/credentials.py): API keys are Secret rows via the normal secret_manager (encrypt_secret/decrypt_secret), name convention ai:{provider}:api_key:{user_id}:{credential_id}. Rotation mutates Secret.encrypted_value in place (id stable).

byok.get_configured_provider(db, user_id, provider, *, surface, model) — model resolution order (its own docstring, byok.py:1-31): (1) explicit model= argument → (2) credential row's default_model → (3) provider class surface_models[surface] if it's in the user's curated models list → (4) first entry of the user's curated models list → (5) provider class surface_models[surface] (plain fallback) → (6) provider class default_model (terminal fallback).

Env-var fallback when no credential row exists — _PROVIDER_ENV_VARS (byok.py): anthropic→ANTHROPIC_API_KEY, openai→OPENAI_API_KEY, google→GEMINI_API_KEY/GOOGLE_API_KEY, groq→GROQ_API_KEY, openrouter→OPENROUTER_API_KEY, ollama→none (never needs a key — unconfigured until a row exists, configured only once one is saved). get_configured_provider raises ProviderNotConfiguredError only when all of: no credential row, no env var, no class default_api_base, and provider != "ollama".

Rate limiting (ai/scheduler.py) — opt-in, not automatic. Per-provider sliding-window RPM/RPD from FLOWFILE_AI_<PROVIDER>_RPM / _RPD (unset ⇒ unenforced; ollama is always in _UNLIMITED_PROVIDERS, never limited). Honors Retry-After on 429; exponential backoff 2s, 4s, 8s, 16s, max 4 retries, ±25% jitter — server Retry-After always wins when longer. byok.get_configured_provider is NOT auto-wrapped; callers opt in via with_provider_retry(provider, ...).

Load-bearing caveat: state is per-process, in-memory (plain deques, no persistence, nothing shared across workers). Under gunicorn -w N (or any multi-worker deploy), the effective aggregate budget is ≈ N × the configured RPM/RPD — each worker enforces its own window independently. For a hard global ceiling, pin to a single worker or divide the configured limit by the worker count. No audit writes happen on retries ("failure is free" — retries are invisible to user-facing quota counters).

---

5. Prompt-log debugging runbook

The prompt log is the "what did the model actually see and say" hatch — prefer it over guessing from output when an agent loops, hallucinates a tool name, references a nonexistent column, or otherwise misbehaves.

1. Set FLOWFILE_AI_LOG_PROMPTS=true in the environment (accepts true/1/yes/on) and re-run the failing flow/chat/agent session.
2. Tail the latest entries: python -m flowfile_core.ai.prompt_log tail 20.
3. Or search: python -m flowfile_core.ai.prompt_log grep PATTERN [SURFACE] — regex over the serialized JSON line, optionally scoped to one AI surface (e.g. agent_staged).
4. Or open the file directly: {storage.base_directory}/ai_prompts/YYYY-MM-DD.jsonl — ~/.flowfile/ai_prompts/... locally, $FLOWFILE_STORAGE_DIR/ai_prompts/... when set, or /app/internal_storage/ai_prompts/... in Docker. One jq-parseable JSON line per LLM round-trip.

Gotcha — tail/grep only read today's file. Both tail(n, date=None) and grep(pattern, surface=None, date=None) default date to datetime.now(timezone.utc).date().isoformat() — UTC, not local time — and the CLI (_cli_main in prompt_log.py) never exposes a date flag. To inspect a prior day, either run just after UTC midnight rollover expecting "yesterday", or drop into Python and call tail(20, date="2026-07-02") / grep(pattern, date="2026-07-02") directly.

Truncation (keeps every line jq-parseable regardless of agent-loop depth): entries over MAX_MESSAGES_BYTES = 256 * 1024 (256 KiB) get their older message bodies replaced with [...truncated, len=N chars] and truncated: true set, while the system prompt plus the most-recent KEEP_RECENT_TURNS = 5 messages are kept verbatim.

Sharing transcripts externally: set FLOWFILE_AI_LOG_PROMPTS_SCRUB=true first. Scrub masks user and tool message bodies only through the regex PII scrubber in ai/safety.py — system and assistant content stay verbatim, because that's the half you actually need to debug compliance against the prompt. Both flags default off; production stays silent unless an operator opts in.

Failure isolation: a prompt-log write failure (disk full, serialization bug) never crashes the LLM call — the wrapper in _litellm_base.py swallows and warns.

---

6. Local model (ai/local_model/manager.py)

On-demand llama-server (llama.cpp) + a GGUF model, downloaded only when the user opts in via /ai/local-model/* routes — the download directory (storage.local_model_directory, ~/.flowfile/local_model) is deliberately excluded from shared.storage_config's _ensure_directories() so it's never created for users who never touch the feature.

• Pinned llama.cpp build: LLAMACPP_BUILD = "b9305" from ggml-org/llama.cpp releases (platform-specific archive per OS/arch). Default model: Qwen3.5 4B Q4_K_M (qwen3.5-4b, ~3 GB, bartowski/Qwen_Qwen3.5-4B-GGUF); the catalog also offers Qwen3.5 2B and 9B. The Qwen2.5-Coder 1.5B/3B and Qwen2.5 7B entries are ModelSpec.legacy=True: hidden from status() unless installed (or selected), still selectable and deletable, so an existing download never becomes invisible or stuck (installed_model_ids and the per-model delete walk the full MODELS).
• _spawn always passes --jinja --reasoning off (thinking off for every surface at once — chat, Simple build, the small-max_tokens JSON surfaces; --reasoning-budget 0 does not stop Qwen3.5 from reasoning, verified on the real 4B) plus the model's ModelSpec.server_args (Qwen3.5 gets Qwen's non-thinking sampling --temp 0.7 --top-p 0.8 --top-k 20). The context ceiling is per model (ModelSpec.max_ctx: 32k legacy, 128k Qwen3.5); get_ctx_size/set_ctx_size clamp to the selected model's value. oneshot.extract_flow_json strips a <think>…</think> block before the candidate scan.
• On-device prompt: every assist-level route (chat_routes — stream and preview share _build_chat_messages — run_failure_routes, docgen_routes, lineage_routes, inline_action_routes) passes local=(provider == LOCAL_PROVIDER_ID) to render_prompt_context / assemble_system_prompt. With local=True the assist suffix is prompts/local_assist.md (no "say do it / switch to agent mode" footer, which a 3B model parrots verbatim and which points at a mode on-device lacks) and the node reference collapses to one line per node (_render_compact_node_reference: ~2.7k tokens for the whole explain prompt instead of ~10k). local=False is the cloud prompt byte for byte (test_context.py pins both). _chat_mode_footer_override returns None for the local provider — the cloud footer text is untouched because intent_router.py matches its wording. Chat forwards client mentions as parsed Mentions, never as raw text (raw text would render a fake ## User request line).
• Simple build writes code by default (local_model/code_build.py, §10): POST /ai/generate with mode="code" asks for a FlowFrame script and the notebook's exec-free interpreter turns it into nodes. The JSON simple / one_shot modes stay reachable and simple stays the request default.
• Module-level singleton — at most one server runs at a time; _lock guards it. LocalProvider resolves the live port lazily on the first .stream() call so constructing it stays synchronous/non-blocking.
• Shut down from core's FastAPI lifespan (main.py's _shutdown_local_model) so a killed core process doesn't orphan the llama-server subprocess.
• Kept out of providers.PROVIDERS (see §4) — never appears in BYOK CRUD, and the tool-calling agent route rejects it outright.

---

7. How to extend safely

• New vendor provider: add a config-only subclass under ai/providers/ (mirror anthropic.py/groq.py — name, default_model, model_prefix, surface_models, default_api_base), register it in registry.PROVIDERS, and add its API-key env var(s) to byok._PROVIDER_ENV_VARS. Never import litellm at module scope — the vendor subclass inherits the lazy _lazy_litellm() call from LiteLLMProvider.
• New AI route: put it under the /ai router (ai/routes.py) so it inherits require_ai_enabled for free — don't hand-roll the 503 check. Add auth per-route (Depends(get_current_active_user), or get_current_admin_user for admin-only) — the feature flag and auth are orthogonal gates, both required.
• New agent tool: the tool catalog auto-derives from the same Pydantic node-settings classes used by the visual graph (see flowfile-node-development) — get the node's settings schema right first and most of the catalog wiring falls out for free. Route it through execute_tool_call in ai/tools/executor/; don't bypass the dispatch seam.
• New prompt content: see the doctrine in §8 before editing anything under ai/prompts/.
• A tool call keeps arriving in a wrong-but-fixable shape: see the doctrine in §9 before touching the prompt or the Pydantic schema.

---