test Feature/advanced observability

DOCUMENTATION.md

@@ -0,0 +1,282 @@
+# Flowk Official Documentation
+
+Welcome to the definitive guide for **Flowk**, an enterprise-grade, statically-verifiable Workflow Orchestrator designed exclusively for pythonic AI agents and LLM pipelines.
+
+This document details every major feature, the underlying API surface, architectural mechanics, and best practices for scaling from prototype to distributed production.
+
+---
+
+## Table of Contents
+1. [Core Architecture & Nodes](#1-core-architecture--nodes)
+2. [State Management & Validation](#2-state-management--validation)
+3. [Asynchronous Execution & Streams](#3-asynchronous-execution--streams)
+4. [Conditional Routing (Standard vs. LLM)](#4-conditional-routing-standard-vs-llm)
+5. [Interrupts & Human-in-the-loop](#5-interrupts--human-in-the-loop)
+6. [Resilience (Retries, Memory, SQLite, Redis)](#6-resilience-retries-memory-sqlite-redis)
+7. [Sub-Graphs & Multi-Agent Composition](#7-sub-graphs--multi-agent-composition)
+8. [Observability (CLI, Time-Travel, Dashboard)](#8-observability-cli-time-travel-dashboard)
+9. [1-Click API Deployment (FastAPI)](#9-1-click-api-deployment-fastapi)
+10. [Plugins & Cost Telemetry](#10-plugins--cost-telemetry)
+
+---
+
+## 1. Core Architecture & Nodes
+
+### The `Graph`
+The `Graph` object acts as the centralized coordinator mapping Nodes together via a Directed Acyclic logic. It automatically detects the first connected node as the `entrypoint`.
+
+```python
+from flowk import Graph
+g = Graph()
+```
+
+### `@g.node()` Customization
+Nodes are asynchronous or synchronous Python functions wrapped inside `@g.node()`. You can inject execution resiliency configurations seamlessly:
+
+- `retries`: Integer amount of times to retry on `NodeExecutionError`
+- `timeout`: Seconds until Node forcibly raises a `TimeoutError`
+- `fallback`: In the event of a catastrophic failure across all retries, a function that returns a default output.
+
+```python
+def fallback_fn(context: str, state: dict) -> str:
+    return "API failed. Outputting generic response."
+
+@g.node(retries=3, timeout=5.0, fallback=fallback_fn)
+async def external_api_call(query: str, state: dict):
+    import aiohttp
+    ...
+```
+
+### `.connect()` API
+Connections strictly chain operations sequentially. If you `.connect(A, B)` and `.connect(A, C)`, Flowk recognizes a diverging logic path and resolves it gracefully:
+- **Async Environment**: Resolves `B` and `C` absolutely concurrently in parallel (`asyncio.gather`).
+- **Sync Environment**: Resolves sequentially based on insertion order.
+
+---
+
+## 2. State Management & Validation
+
+Instead of passing massive arguments between nodes manually, Flowk leverages an implicit shared `dict` space bound to execution transactions. 
+
+### Pydantic Strict Evaluation
+To enforce zero data corruption across massive AI pipelines, specify a typed schema via `state_schema` on creation. **Flowk validates the mutable state object across every single node transition.**
+
+```python
+from pydantic import BaseModel
+
+class EnterpriseState(BaseModel):
+    user_id: int
+    authorized: bool = False
+    context_window: list = []
+
+g = Graph(state_schema=EnterpriseState)
+
+@g.node()
+def modify_agent(input_data: Any, state: dict):
+    # Appending cleanly
+    state["context_window"].append(input_data)
+
+    # If a developer accidentally writes:
+    # state["authorized"] = "yes"
+    # Flowk will raise a strict GraphExecutionError and kill the flow before the next node executes!
+```
+
+---
+
+## 3. Asynchronous Execution & Streams
+
+### Awaiting Graph Outcomes (`arun`)
+Flowk was built for scalable serverless and WebSocket deployment. Standard `g.run()` executes the event-loop blocker, but `g.arun()` operates entirely asynchronously:
+
+```python
+result = await g.arun(
+    input_data="query", 
+    session_id="user_123", 
+    initial_state={"authorized": True} # Scoped state injection
+)
+```
+
+### Event Yielding (`astream`)
+Perfect for Server-Sent Events (SSE) or UI loader updates. `g.astream(...)` evaluates the pipeline normally, but immediately `yields` the outcome dictionary generated by exactly the node that just finished.
+
+```python
+async for event in g.astream(input_data):
+    node_source = event["source"] 
+    node_result = event["output"]
+    print(f"[{node_source}]: {node_result}")
+    # Perfect for feeding into a front-end React interface incrementally
+```
+
+---
+
+## 4. Conditional Routing (Standard vs. LLM)
+
+### Standard Code-Based Routers (`g.route()`)
+When paths diverge based on business logic, use `.route()` with a deterministic mapping dictionary:
+
+```python
+def router_logic(input_data: str):
+    return "fast" if len(input_data) < 10 else "complex"
+
+fast_node = ... 
+complex_node = ...
+
+router = g.route(router_logic, {
+    "fast": fast_node,
+    "complex": complex_node
+})
+
+g.connect(parser, router) # Connect parent to router
+```
+
+### Zero-Boilerplate Intelligent Auto-Routing (`@g.llm_router`)
+Replaces thousands of lines of fragile NLP evaluation by passing the branch decision directly to an LLM using few-shot classification boundaries.
+
+```python
+@g.llm_router(
+    model="gpt-4o-mini", # Requires OPENAI_API_KEY exported
+    targets={
+        "sentiment_node": "Executes sentiment and mood analysis.",
+        "query_node": "Executes factual web searches."
+    }
+)
+def intelligent_supervisor(state: dict):
+    # 1. Provide the string block you want the LLM to base its decision on.
+    return state["raw_text"]
+
+# Flowk automatically passes the targets dict and the returned state data to the LLM, 
+# strictly validating that the LLM returns exactly "sentiment_node" or "query_node", 
+# and seamlessly resolves the pipeline down the chosen branch.
+```
+
+---
+
+## 5. Interrupts & Human-in-the-loop
+
+Long-running pipelines frequently require external confirmations (e.g. paying an invoice or pushing codebase edits).
+
+```python
+# 1. Halt execution definitively before hitting a sensitive node
+g.compile(interrupt_before=["commit_action"])
+
+session_id = "process-tx-999"
+
+# 2. Run the graph normally. 
+# It will yield a `type: 'interrupt'` event and terminate exactly before the commit.
+async for event in g.astream(input_data, session_id=session_id):
+    if event.get("type") == "interrupt":
+         print("Suspended! Waiting for human review...")
+
+# ... Hours or days later...
+
+# 3. Resume! Provide the exact identical session_id and execution 
+# magically continues perfectly as if it was never stopped
+await g.arun(None, session_id=session_id)
+```
+
+---
+
+## 6. Resilience (Retries, Memory, SQLite, Redis)
+
+Flowk preserves your AI run threads natively. When deploying, assign `checkpoint_db` dynamically:
+
+```python
+# InMemory Dict (Great for Pytest)
+g = Graph() 
+
+# Local SQLite (Great for local applications / Streamlit)
+g = Graph(checkpoint_db="local_flow.db")
+
+# Distributed Redis (Great for horizontally scaled load-balancers)
+# ** Requires pip install flowk[redis]
+g = Graph(checkpoint_db="redis://localhost:6379/0") 
+```
+
+All standard operations `run`, `arun`, and `astream` support executing on an explicit `session_id`. If `session_id` is supplied, state histories are queried locally or over the network automatically logic-gated to that ID.
+
+---
+
+## 7. Sub-Graphs & Multi-Agent Composition
+
+The crown jewel of agentic scale: embedding fully compiled workflow graphs *as executable nodes* inside a larger parent graph.
+
+```python
+# 1. Sub-Agent Graph (Research)
+sub = Graph()
+sub.connect(fetch_data, summarize)
+
+# 2. Mount it as a Node 
+# Extracts data from the parent's `research_meta` key as its input bounds, 
+# and dumps its output perfectly back into that key.
+research_agent_node = sub.as_node(state_key="research_meta")
+
+# 3. Connect to Parent pipeline
+main = Graph()
+main.connect(draft_outline, research_agent_node)
+```
+This isolates complex pipelines without polluting supervisor-level logic domains.
+
+---
+
+## 8. Observability (CLI, Time-Travel, Dashboard)
+
+### Flowk UI (Local Dashboard)
+Inspect run histories, trace logic changes, and debug global state permutations explicitly via our Streamlit UI.
+```bash
+flowk ui
+# Operates at http://localhost:8501 pulling from SQLite local checkpoints
+```
+
+### CLI Operations
+Render beautiful flow charts to your developer terminal before production to visually confirm structural intent:
+```python
+g.show()
+# Automatically DFS parses nodes to render ascii-art branching topologies.
+```
+
+### Time Travel Metrics
+Debug a production failure identically:
+```python
+# 1. Run noisily in the terminal
+g.debug("hello_world", session_id="abc-999")
+
+# 2. Capture a Run_ID from the logger trace (e.g. 'run-xxx-yyy-zzz')
+# 3. Execute time travel playback structurally reproducing the exact data trace 
+# input/outputs for every single nested node exactly as they occurred historically.
+g.replay('run-xxx-yyy-zzz')
+```
+
+---
+
+## 9. 1-Click API Deployment (FastAPI)
+
+Flowk dynamically evaluates your compiled graphs schema bounds, translates it to Pydantic OpenAPI requirements, and mounts standard HTTP pipelines.
+
+```python
+# Mount the Graph exactly as-is into an ASGI process
+if __name__ == "__main__":
+    g.serve(host="0.0.0.0", port=8000)
+```
+
+Provides Auto-Generated OpenTelemetry routes:
+- `POST /invoke`: Blocks until final resolution, returning the exact output text trace.
+- `POST /stream`: Keeps connection open serving chunks of identical SSE formatted event data per node execution (`astream` equivalent). Contains `data:` prefixes mapping to standard HTTP specifications.
+
+---
+
+## 10. Plugins & Cost Telemetry
+
+Want to track precisely how much an agent pipeline costs over a 12-hour period? Mount standard flowk plugins globally to intercept events.
+
+```python
+from flowk import MetricsRegistry
+from flowk.plugins.llm import OpenAIPlugin, AnthropicPlugin
+from flowk.plugins.base import PluginManager
+
+# Mount global cost estimators listening to LLM outputs
+PluginManager.register(OpenAIPlugin(model="gpt-4o-mini")) 
+PluginManager.register(AnthropicPlugin(model="claude-3-opus"))
+
+print(MetricsRegistry.get_summary())
+# Resolves structured breakdown of exact Tokens input/output globally and $ estimated scale.
+```