Skip to main content

FluxLoop SDK

The FluxLoop SDK provides Python decorators and utilities for instrumenting AI agents with automatic tracing and observation collection.

Version​

Current version: 0.1.7

Installation​

pip install fluxloop

Requirements​

  • Python 3.11 or higher
  • pip
Python Version Requirement

The FluxLoop SDK requires Python 3.11+ due to advanced type hints and protocol features.

Features​

  • 🔍 Automatic Tracing: Instrument your agent code with simple decorators
  • 📊 Rich Context: Capture inputs, outputs, metadata, and nested observations
  • 🔄 Async Support: First-class support for both sync and async functions
  • 🎯 Framework Integration: Built-in support for LangChain, LangGraph, and custom frameworks
  • 📝 Observation Streaming: Real-time event collection with hierarchical structure
  • 💾 Flexible Storage: File-based (JSONL) or HTTP collector backends
  • 🔧 Recording Mode: Capture and replay complex function arguments
  • 🧪 Test Integration: Seamless pytest integration for agent testing

Quick Example​

import fluxloop

# Initialize (optional - uses sensible defaults)
fluxloop.configure(
    storage="file",
    base_path="./traces"
)

# Decorate your agent function
@fluxloop.agent(name="ChatAgent")
def my_agent(prompt: str) -> str:
    """Your agent logic here."""
    # Agent implementation
    result = process_prompt(prompt)
    return result

# Run it - traces are automatically collected
response = my_agent("Hello, world!")

Core Components​

Decorators​

FluxLoop provides specialized decorators for different observation types:

DecoratorPurposeObservation Type
@fluxloop.agent()Agent entry pointsAGENT
@fluxloop.trace()General function tracingSPAN (configurable)
@fluxloop.tool()Tool/function callsTOOL
@fluxloop.prompt()LLM completionsGENERATION
@fluxloop.instrument()Context manager for scopesConfigurable

Client & Configuration​

  • fluxloop.configure() - Global configuration (storage, debug, recording)
  • fluxloop.FluxLoopClient() - Advanced client for custom setups
  • fluxloop.load_env() - Load configuration from .env files

Context Management​

  • get_current_context() - Access current trace context
  • get_current_trace() - Get current trace metadata
  • Manual observation push/pop for fine-grained control

Storage Backends​

  • FileStorage - Local JSONL files (default)
    • traces_YYYYMMDD.jsonl - Trace metadata
    • observations_YYYYMMDD.jsonl - Observation stream
  • HTTPStorage - Remote collector service
    • Real-time streaming to FluxLoop collector
    • Buffered batch uploads

Advanced Features​

Recording Mode​

Capture complex function arguments for replay in experiments:

import fluxloop

# Enable recording
fluxloop.configure(
    record_args=True,
    recording_file="recordings/args.jsonl"
)

# Arguments are automatically captured
@fluxloop.agent()
def my_handler(request: ComplexRequest):
    return process(request)

Async & Streaming Support​

Full support for async functions and async generators:

@fluxloop.agent()
async def async_agent(prompt: str) -> str:
    result = await process_async(prompt)
    return result

@fluxloop.prompt(model="gpt-4o")
async def stream_response(messages: list):
    """Stream LLM chunks."""
    async for chunk in llm.stream(messages):
        yield chunk  # Each chunk traced

Framework Integration​

Safe integration with external frameworks (LangChain, LangGraph, custom agents):

# Pattern A: Manual instrumentation (safest)
from fluxloop import get_current_context
from fluxloop.models import ObservationData, ObservationType

async def my_tool(param: str) -> dict:
    fl_ctx = get_current_context()
    obs = None
    
    if fl_ctx and fl_ctx.is_enabled():
        obs = ObservationData(
            type=ObservationType.TOOL,
            name="my_tool",
            input={"args": {"param": param}},
        )
        fl_ctx.push_observation(obs)
    
    try:
        result = process(param)
        if obs:
            obs.output = result
        return result
    finally:
        if fl_ctx and obs:
            fl_ctx.pop_observation()

# Pattern B: Decorator stacking (framework outermost)
@framework_tool_decorator(...)
@fluxloop.tool(name="my_tool")
async def my_tool(...):
    ...

See Custom Framework Integration for detailed guidance.

Observation Types​

FluxLoop captures different types of observations:

TypeDescriptionUse Case
AGENTAgent-level operationTop-level agent entry points
SPANGeneric timing spanGeneral function tracing
TOOLTool/function callExternal tools, APIs, functions
GENERATIONLLM completionModel inference, prompts
RETRIEVALDocument retrievalRAG, database queries
EMBEDDINGEmbedding generationVector generation
EVENTPoint-in-time eventLogging, milestones

Output Structure​

Traces and observations are saved in JSONL format:

./traces/
├── traces_20250117.jsonl         # Trace metadata
└── observations_20250117.jsonl    # Observation stream

Each trace contains:

  • trace_id: Unique identifier
  • session_id: Session grouping
  • start_time, end_time: Timestamps
  • metadata: Custom metadata
  • tags: Categorization tags
  • status: Success/error status

Each observation contains:

  • id, trace_id: Identifiers
  • type: Observation type (AGENT, TOOL, etc.)
  • name: Human-readable name
  • input, output: Captured data
  • start_time, end_time: Timing
  • parent_id: For nested observations
  • metadata: Additional context

Integration with FluxLoop CLI​

The SDK works seamlessly with FluxLoop CLI for experiments:

# configs/simulation.yaml
runner:
  target: "my_module:my_agent"  # Your decorated function
  iterations: 10
# Run tests locally
fluxloop test

# Upload results to Web Platform
fluxloop sync upload

See CLI Integration for details.

What's Next?​

Getting Started​

Configuration​

API Reference​

Framework Integration​

Need help? Check the API Reference or GitHub Issues.