Observability

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.

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.

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

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.

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"}],
)

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"}],
)

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",
)

from langchain_openai import ChatOpenAI
from fluiq import instrument

instrument(api_key="fl_...")

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

from fluiq import instrument

instrument(api_key="fl_...")

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

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.

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, LangGraph, CrewAI, and Google ADK agents are traced automatically without a decorator; the integration emits a root span for the run and child spans for every internal step.

Fan-out / fan-in orchestrations are captured as a true graph, not a flat list. When several agents run in parallel and their outputs join into a single downstream step, Fluiq detects the real dependency edges and renders the run as a DAG in the trace drawer, with fan-out, joins, and loop-backs included.

• LangGraph: join (fan-in) nodes are resolved from the graph's declared edges captured at compile(), so multi-parent steps are exact regardless of the trigger encoding. Each run shows in the Agents view as LangGraph(node_a, node_b, …), named after its nodes.
• CrewAI: task dependencies (task.context) become graph edges; a crew shows as Crew(agent_1, agent_2, …).
• Google ADK: {state_key} placeholders in an agent's instruction are resolved to the upstream agents that produced them.

For custom orchestrations, declare joins yourself with fluiq.join_parents(...) so a step that consumes multiple upstream results is rendered with all of its parents.

Multimodal steps are preserved too: image, audio, and file parts are kept as lightweight media references (kind, mime, size, and a content hash) on the trace without bloating storage.

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.