FlowgentraAI

Build AI agent workflows with graphs -- Python bindings for the FlowgentraAI Rust engine, powered by PyO3.

Installation

pip install flowgentra-ai

From source (development)

cd flowgentra-ai-py
pip install maturin
maturin develop

Quick Start

Build a graph workflow (recommended)

The primary way to use FlowgentraAI in Python is the StateGraph API. Define nodes as Python functions, wire them with edges, compile, and invoke():

from flowgentra_ai.graph import StateGraph, END
from typing import TypedDict

# 1. Define state shape — TypedDict gives IDE autocomplete
class GreetState(TypedDict):
    name: str
    greeting: str

# 2. Define node functions (receive and return state dict)
def greet(state: GreetState) -> GreetState:
    return {**state, "greeting": f"Hello, {state['name']}!"}

def uppercase(state: GreetState) -> GreetState:
    return {**state, "greeting": state["greeting"].upper()}

# 3. Build the graph
builder = StateGraph(GreetState)
builder.add_node("greet", greet)
builder.add_node("uppercase", uppercase)
builder.set_entry_point("greet")
builder.add_edge("greet", "uppercase")
builder.add_edge("uppercase", END)
graph = builder.compile()

# 4. Invoke with initial state (plain dict)
result = graph.invoke({"name": "World", "greeting": ""})
print(result)
# {"name": "World", "greeting": "HELLO, WORLD!"}

Config-driven agent

For YAML-configured agents with auto-discovered handlers:

from flowgentra_ai.agent import Agent

agent = Agent.from_config_path("config.yaml")
agent.state["input"] = "Hello, world!"
result = agent.run()
print(result.to_dict())

Core Concepts

State

State is a thread-safe dict-like container that flows through your graph:

from flowgentra_ai import State

state = State({"key": "value"})
state["name"] = "FlowgentraAI"
print(state["name"])        # "FlowgentraAI"
print("name" in state)      # True
print(state.to_dict())      # {"key": "value", "name": "FlowgentraAI"}
print(len(state))           # 2

# Serialization
json_str = state.to_json()
state = State.from_json('{"a": 1}')
state = State.from_dict({"a": 1})

Conditional Routing

Route dynamically based on state:

from flowgentra_ai.graph import StateGraph, END
from typing import TypedDict

class RouterState(TypedDict):
    input: str
    category: str
    output: str

def classify(state: RouterState) -> RouterState:
    category = "greeting" if "hello" in state["input"].lower() else "question"
    return {**state, "category": category}

def handle_greeting(state: RouterState) -> RouterState:
    return {**state, "output": "Hi there!"}

def handle_question(state: RouterState) -> RouterState:
    return {**state, "output": "Let me think about that..."}

def router(state: RouterState) -> str:
    return "handle_greeting" if state["category"] == "greeting" else "handle_question"

builder = StateGraph(RouterState)
builder.add_node("classify", classify)
builder.add_node("handle_greeting", handle_greeting)
builder.add_node("handle_question", handle_question)
builder.set_entry_point("classify")
builder.add_conditional_edge("classify", router)
builder.add_edge("handle_greeting", END)
builder.add_edge("handle_question", END)
graph = builder.compile()

result = graph.invoke({"input": "hello world", "category": "", "output": ""})
print(result["output"])  # "Hi there!"

LLM

Call LLM providers directly:

from flowgentra_ai.llm import LLM, Message

# Preferred: construct directly (key auto-resolved from env var)
client = LLM(provider="openai", model="gpt-4o")

# Or pass an explicit key
client = LLM(provider="anthropic", model="claude-3-5-haiku-20241022", api_key="sk-ant-...")

# Simple chat
response = client.chat([
    Message.system("You are a helpful assistant."),
    Message.user("What is Rust?"),
])
print(response.content)

# With token usage tracking
response, usage = client.chat_with_usage([Message.user("Hello!")])
if usage:
    print(f"Tokens: {usage.total_tokens}, Cost: ${usage.estimated_cost('gpt-4o'):.4f}")

# With retry and caching
robust_client = client.with_retry(max_retries=3).cached(max_entries=100)

Multi-Agent Supervisor

Orchestrate multiple agent graphs with 11 strategies:

from flowgentra_ai.supervision import (
    Supervisor, SupervisorNodeConfig, OrchestrationStrategy,
    ParallelMergeStrategy,
)
from flowgentra_ai import State

# Simple mode (router function)
def router(state):
    if "research" in (state.get_string("task") or ""):
        return "researcher"
    return "FINISH"

sup = Supervisor(router)
sup.add_agent("researcher", research_graph)
result = sup.run(State({"task": "research AI trends"}))

# Strategy mode (parallel, sequential, retry_fallback, debate, etc.)
config = SupervisorNodeConfig(
    "coordinator",
    ["researcher", "writer"],
    OrchestrationStrategy.parallel(),
)
config.set_child_timeout_ms(30000)
config.set_merge_strategy(ParallelMergeStrategy.latest())

sup = Supervisor.from_config(config)
sup.add_agent("researcher", research_graph)
sup.add_agent("writer", writer_graph)
result = sup.run(State({"task": "write article"}))

Strategies: sequential, parallel, autonomous, dynamic, round_robin, hierarchical, broadcast, map_reduce, conditional_routing, retry_fallback, debate.

Built-in Node Types

from flowgentra_ai.graph import StateGraph
from typing import TypedDict

class State(TypedDict):
    query: str
    response: str

def fetch_fn(state: dict) -> dict: ...
def slow_fn(state: dict) -> dict: ...
def refine_fn(state: dict) -> dict: ...
client = ...  # an LLM instance

builder = StateGraph(State)

# Retry with backoff
builder.add_retry_node("fetch", fetch_fn, max_retries=3, backoff_ms=1000)

# Timeout enforcement
builder.add_timeout_node("slow_op", slow_fn, timeout_ms=5000)

# LLM-integrated node
builder.add_llm_node("generate", client, prompt_key="query", output_key="response")

# LLM-driven planner (dynamic routing)
builder.add_planner_node("planner", client)

# Iterative evaluation/refinement
from flowgentra_ai.evaluation import EvaluationNodeConfig
config = EvaluationNodeConfig("refine", field_state="draft", min_confidence=0.8)
builder.add_evaluation_node(handler=refine_fn, config=config)

Conversation Memory

Track multi-turn conversations:

from flowgentra_ai.memory import ConversationMemory
from flowgentra_ai.llm import Message

mem = ConversationMemory(max_messages=100)
mem.add_message("thread-1", Message.user("Hello"))
mem.add_message("thread-1", Message.assistant("Hi! How can I help?"))

messages = mem.messages("thread-1")
print(len(messages))  # 2

Or use the high-level MemoryAwareAgent:

from flowgentra_ai.agent import MemoryAwareAgent

agent = MemoryAwareAgent.from_config("config.yaml")
agent.set_thread_id("user_123")
answer = agent.run_turn("What is Rust?")
answer2 = agent.run_turn("What are its main features?")  # remembers context

RAG (Retrieval-Augmented Generation)

from flowgentra_ai.rag import (
    Document, Embeddings, InMemoryVectorStore,
    Retriever, RetrievalConfig,
)
from flowgentra_ai.types import chunk_text, extract_text, estimate_tokens

# Text utilities
chunks = chunk_text("long text...", chunk_size=500, overlap=50)
text = extract_text("document.pdf")  # needs an actual file on disk
tokens = estimate_tokens("some text")

# Vector store + retrieval
emb = Embeddings.openai("sk-...", "text-embedding-3-small")
store = InMemoryVectorStore()

doc = Document("doc1", "Rust is a systems programming language.")
store.index(doc, emb.embed(doc.text))

config = RetrievalConfig.semantic(top_k=5, threshold=0.7)
retriever = Retriever(store, emb, config)
results = retriever.retrieve("What is Rust?")

for r in results:
    print(f"{r.id}: {r.text} (score: {r.score:.3f})")

Prebuilt Agents

Use ready-made agent patterns:

from flowgentra_ai.agent import ZeroShotReAct, ToolSpec
from flowgentra_ai.llm import LLM

tool = ToolSpec("search", "Search the web")
tool.add_parameter("query", "string")
tool.set_required("query")

agent = ZeroShotReAct(
    name="my_agent",
    llm=LLM(provider="openai", model="gpt-4o"),
    system_prompt="You are a helpful assistant.",
    tools=[tool],
)

result = agent.run("What is the weather today?")
print(result)

Human-in-the-Loop

Interrupt graph execution for human review:

from flowgentra_ai.graph import StateGraph, END
from typing import TypedDict

class PublishState(TypedDict):
    topic: str
    draft: str
    approved: bool

def draft_fn(state: dict) -> dict:
    return {"draft": f"An article about {state['topic']}"}

def publish_fn(state: dict) -> dict:
    return {"approved": True}

builder = StateGraph(PublishState)
builder.add_node("draft", draft_fn)
builder.add_node("publish", publish_fn)
builder.set_entry_point("draft")
builder.add_edge("draft", "publish")
builder.add_edge("publish", END)
builder.interrupt_before("publish")  # pause before publishing
builder.set_checkpointer("./checkpoints")
graph = builder.compile()

# First run -- raises AgentExecutionError at the breakpoint before "publish"
try:
    graph.invoke_with_thread("thread-1", {"topic": "AI", "draft": "", "approved": False})
except Exception:
    pass  # paused at the breakpoint; state is checkpointed under "thread-1"

# Human reviews, then resumes
result = graph.resume("thread-1")

# Or resume with modified state
result = graph.resume_with_state("thread-1", {"topic": "AI", "draft": "", "approved": True})

Visualization

Export graphs as DOT or Mermaid diagrams:

graph = builder.compile()
print(graph.to_mermaid())
print(graph.to_dot())
print(graph.node_names())

Tool Registry

Use built-in tools or register custom ones:

from flowgentra_ai.tools import ToolRegistry

registry = ToolRegistry.with_builtins()
print(registry.list_names())  # ["calculator", "search", ...]

result = registry.call_tool("calculator", {"operation": "add", "a": 2, "b": 3})
print(result)  # 5

Model Pricing

from flowgentra_ai.types import model_pricing
from flowgentra_ai.llm import TokenUsage

pricing = model_pricing("gpt-4")  # (input_price, output_price) per million tokens
usage = TokenUsage(1000, 500)
cost = usage.estimated_cost("gpt-4")

Supported LLM Providers

Full Documentation

For complete API reference and guides, visit the documentation site.

License

MIT