flowgentra-ai 0.3.2

Python bindings for FlowgentraAI - build AI agents with graphs

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

Provider	Config name
OpenAI	"openai"
Anthropic	"anthropic"
Mistral	"mistral"
Groq	"groq"
Ollama (local)	"ollama"
HuggingFace	"huggingface"
Azure OpenAI	"azure"

Full Documentation

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

License

MIT