Observability

Fluiq captures every LLM call, tool invocation, and retrieval step automatically — including model, messages, response, latency, token usage, and cost. Use the @trace decorator to group any Python function into the same trace tree.

@trace decorator

Wrap any function with @trace to record inputs, outputs, latency, and errors. Async functions are detected and awaited automatically. Nested calls preserve parent / child relationships.

Python

from fluiq import instrument, trace

instrument(api_key="fl_...")

@trace
def retrieve(question: str) -> list[str]:
    return vector_store.similarity_search(question, k=4)

@trace
async def answer(question: str) -> str:
    docs = retrieve(question)                 # nested span
    return await llm.ainvoke(prompt(question, docs))

Pass name= to override the function name used as the agent identity on the dashboard.

Python

@trace(name="research_agent")
def run(question: str) -> str:
    ...

Fail-open by design

Every span emission is wrapped in a safety guard so a Fluiq SDK error never crashes your application. Network failures, malformed payloads, missing optional dependencies, and dashboard outages are absorbed silently.

Auto-instrumentation

instrument() patches every supported provider it can find on import. If a provider isn't installed the patch is skipped silently — you never need feature flags.

OpenAI

Patches chat completions, responses, parse, streaming, embeddings, images, and audio — sync and async.

Python

import openai
from fluiq import instrument

instrument(api_key="fl_...")

client = openai.OpenAI()
client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Anthropic

Patches the Messages API and the Beta client, sync and async.

Python

import anthropic
from fluiq import instrument

instrument(api_key="fl_...")

client = anthropic.Anthropic()
client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=512,
    messages=[{"role": "user", "content": "Hello"}],
)

Gemini & Vertex AI

Patches google-genai and Vertex AI — generation, streaming, and count_tokens, sync and async.

Python

from google import genai
from fluiq import instrument

instrument(api_key="fl_...")

client = genai.Client()
client.models.generate_content(
    model="gemini-2.5-pro",
    contents="Hello",
)

LangChain

Patches the LangChain runtime so chains, agents, and retrievers emit traces automatically.

Python

from langchain_openai import ChatOpenAI
from fluiq import instrument

instrument(api_key="fl_...")

llm = ChatOpenAI(model="gpt-4o")
llm.invoke("Hello")

MCP

Wraps MCP client initialize so tool calls flowing through Model Context Protocol servers are traced.

Python

from fluiq import instrument

instrument(api_key="fl_...")

# Any MCP client.initialize() call is now traced
# alongside the LLM that invokes the tool.

Agents

An agent in Fluiq is any function or chain you want to monitor as a single unit of work. Wrap your entrypoint with @trace so every nested LLM call and tool invocation is grouped under one root — and aggregated as one row on the Agents dashboard.

Python

from fluiq import instrument, trace

instrument(api_key="fl_...")

@trace
def run_research_agent(question: str) -> str:
    plan = planner(question)              # nested @trace
    docs = retrieve(plan)                 # nested @trace
    return synthesize(question, docs)     # nested @trace

LangChain and LangGraph agents are traced automatically without a decorator — the integration emits a root span for the runnable and child spans for every internal step.

Cost analytics

Every traced LLM call is priced server-side using the provider's published rates and rolled up across the full agent run. The Traces table shows per-call cost; the Agents dashboard groups by agent key and reports total cost, avg cost per run, tokens, and latency. Sort by total cost to find your most expensive agents in production.