Structured outputs

Work with an agent's results as data, not just text. Tables, charts, maps, and graphs have structured specifications that your application can inspect, serialize, and render. Their data can resolve on demand as parameter selections change. Use outputs from a chat session or construct them without an agent.

Use outputs from a session

After result = await session.run(...), resolve the returned artifacts and inspect their data and queries:

from tabulaflow.output.resolver import (
    OutputResolver,
    ResolvedChartArtifact,
    ResolvedTableArtifact,
    UnavailableArtifact,
)

resolved = await OutputResolver(session.output_store).resolve(result.output)
for artifact in resolved.artifacts:
    if isinstance(artifact, UnavailableArtifact):
        print("Unavailable:", artifact.artifact_id, artifact.reason)
    elif isinstance(artifact, (ResolvedTableArtifact, ResolvedChartArtifact)):
        print("Artifact:", artifact.label)
        print("Source:", artifact.result.metadata.connector_alias)
        print("SQL:", artifact.result.metadata.query)
        print("DataFrame:\n", artifact.result.df)

Chart artifacts also expose their Vega-Lite specification as artifact.spec. Your frontend can render the resolved artifacts directly.

Example: Warehouse transfer graph

You're reviewing transfers between warehouses. Explore the transfers as an interactive graph without using a graph database, with a filter for the minimum transfer size.

Create the sample database

import pandas as pd

from tabulaflow.data import DataConnectorRegistry, SQLConnector


logistics = await SQLConnector.from_url_async("sqlite+aiosqlite:///:memory:", read_only=False)
registry = DataConnectorRegistry()
# Make the connector available to the output store.
registry.register("logistics", logistics)
await logistics.write_dataframe_async(
    pd.DataFrame(
        columns=["origin", "destination", "units"],
        data=[
            ("Chicago", "Dallas", 500),
            ("Chicago", "Denver", 200),
            ("Dallas", "Austin", 350),
            ("Denver", "Seattle", 80),
        ],
    ),
    "transfers",
)

from tabulaflow.output.store import OutputStore


store = OutputStore(registry=registry)

Add interactive controls

Declare a minimum transfer size and a query that uses it. Creating the source does not execute the query:

from tabulaflow.output.specs import NumberParameter


min_units = NumberParameter(
    id="min_units",
    label="Minimum units transferred",
    min=0,
    max=500,
    step=50,
    default=100,
)
source = store.add_parameterized_artifact_source(
    connector_alias="logistics",
    parameters=[min_units],
    query_template=(
        "SELECT origin, destination, units FROM transfers "
        "WHERE units >= {{ min_units }} ORDER BY origin, destination"
    ),
)

NumberParameter provides the bounds and default for a slider or numeric input. Use ChoiceParameter for a fixed set of options. See parameters and selections.

Resolve data on demand

Share the source between a table and a graph. Origin and destination values become graph nodes; each transfer becomes a directed edge:

from tabulaflow.output.specs import GraphArtifactSpec, OutputSpec, TableArtifactSpec

graph = GraphArtifactSpec(
    id="transfer_graph",
    source_ids=[source.id],
    spec={
        "nodes": [
            {"source_id": source.id, "id": "origin"},
            {"source_id": source.id, "id": "destination"},
        ],
        "edges": [{"source_id": source.id, "source": "origin", "target": "destination", "label": "units"}],
    },
)
output = OutputSpec(
    parameters=[min_units],
    sources=[source],
    artifacts=[TableArtifactSpec(id="transfer_table", source_id=source.id), graph],
)

Change the selection without another model call. Both artifacts share one query result, and returning to an earlier selection reuses that result:

from tabulaflow.output.resolver import OutputResolver, ResolvedGraphArtifact, ResolvedTableArtifact, UnavailableArtifact


resolver = OutputResolver(store)
for threshold in (100, 300, 100):
    resolved = await resolver.resolve(output, {"min_units": threshold})
    print("Selection:", resolved.selection)
    for artifact in resolved.artifacts:
        if isinstance(artifact, UnavailableArtifact):
            print("Unavailable:", artifact.artifact_id, artifact.reason)
        elif isinstance(artifact, ResolvedTableArtifact):
            print("Result ID:", artifact.result.metadata.id)
            print("SQL:", artifact.result.metadata.query)
            print("DataFrame:\n", artifact.result.df)
        elif isinstance(artifact, ResolvedGraphArtifact):
            print("Graph nodes:", [node.id for node in artifact.graph.nodes])
            print("Graph edges:", [(edge.source, edge.target, edge.label) for edge in artifact.graph.edges])

Sample output

Selection: {'min_units': 100}
Result ID: R1
SQL: SELECT origin, destination, units FROM transfers WHERE units >= 100 ORDER BY origin, destination
DataFrame:
     origin destination  units
0  Chicago      Dallas    500
1  Chicago      Denver    200
2   Dallas      Austin    350
Graph nodes: ['Austin', 'Chicago', 'Dallas', 'Denver']
Graph edges: [('Chicago', 'Dallas', '500'), ('Chicago', 'Denver', '200'), ('Dallas', 'Austin', '350')]
Selection: {'min_units': 300}
Result ID: R2
SQL: SELECT origin, destination, units FROM transfers WHERE units >= 300 ORDER BY origin, destination
DataFrame:
     origin destination  units
0  Chicago      Dallas    500
1   Dallas      Austin    350
Graph nodes: ['Austin', 'Chicago', 'Dallas']
Graph edges: [('Chicago', 'Dallas', '500'), ('Dallas', 'Austin', '350')]
Selection: {'min_units': 100}
Result ID: R1
SQL: SELECT origin, destination, units FROM transfers WHERE units >= 100 ORDER BY origin, destination
DataFrame:
     origin destination  units
0  Chicago      Dallas    500
1  Chicago      Denver    200
2   Dallas      Austin    350
Graph nodes: ['Austin', 'Chicago', 'Dallas', 'Denver']
Graph edges: [('Chicago', 'Dallas', '500'), ('Chicago', 'Denver', '200'), ('Dallas', 'Austin', '350')]

After installing TabulaFlow, run the complete example without a database server or API key:

tabulaflow examples run structured-outputs

See result storage for persistence and cache behavior.