Explore Queries and Streaming Answers¶

Explore/Data Asking is not a conventional request-response call. Python analyze_data_stream and Go AnalyzeDataStream return a Server-Sent Events (SSE) stream. The application must keep reading, retain the request_id from the init event, and close the stream on completion or failure.

Request¶

Both SDKs use the same request fields:

JSON field	Required	Purpose
`question`	Yes	The analysis question; the SDK rejects an empty string
`source`	No	An identifier for the calling source
`session_id`	No	Continue an existing session
`session_name`	No	Session name
`config`	No	Data source, scope, filter, and context configuration

See Explore data sources for table and file selection.

Python: consume the complete stream¶

from moi import RawClient

raw = RawClient("https://api.example.com", "your-api-key")

request = {
    "question": "Summarize the renewal conditions.",
    "session_id": session_id,
    "config": {
        "data_source": {
            "type": "specified",
            "files": {
                "type": "specified",
                "file_id_list": selected_file_ids,
            },
        }
    },
}

request_id = None
with raw.analyze_data_stream(request) as stream:
    while True:
        event = stream.read_event()
        if event is None:
            break

        if event.step_type == "init":
            init = event.get_init_event_data()
            if init is not None:
                request_id = init.request_id

        # Use event.type, event.source, event.step_type, event.step_name,
        # and event.data to update the UI or run record.
        handle_event(event)

read_event() returns None at EOF. The context manager calls close(); without with, close the stream in finally.

Go: consume the complete stream¶

package main

import (
    "context"
    "io"

    sdk "github.com/matrixorigin/moi-go-sdk"
)

func analyze(ctx context.Context, client *sdk.RawClient, fileIDs []string) error {
    req := &sdk.DataAnalysisRequest{
        Question: "Summarize the renewal conditions.",
        Config: &sdk.DataAnalysisConfig{
            DataSource: &sdk.DataSource{
                Type: "specified",
                Files: &sdk.FileConfig{
                    Type:       "specified",
                    FileIDList: fileIDs,
                },
            },
        },
    }

    stream, err := client.AnalyzeDataStream(ctx, req)
    if err != nil {
        return err
    }
    defer stream.Close()

    for {
        event, err := stream.ReadEvent()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        if event.StepType == "init" {
            init := event.GetInitEventData()
            if init != nil {
                saveRequestID(init.RequestID)
            }
        }
        handleEvent(event)
    }
}

Go ReadEvent() uses io.EOF for normal stream completion. AnalyzeDataStream accepts CallOption values; use WithStreamBufferSize to change the initial buffer for large events. A configured read timeout applies between messages, not to the total analysis duration.

Event model¶

The stable outer fields of DataAnalysisStreamEvent are:

type, used by classification, completion, error, and other events;
source, with rag and nl2sql documented in source comments;
data, whose shape depends on the event;
step_type and step_name, used by some NL2SQL and step events;
raw_data / RawData, retaining raw JSON for unknown events and diagnosis.

The SDK source documents these events:

Event	Purpose
`init`	First event; `data` contains `request_id` and `session_title`
`classification`	Question classification
`decomposition`	Decomposition of attribution questions
`step_start` / `step_complete`	Attribution step lifecycle
`chunks` / `answer_chunk`	RAG data and incremental answer content
`step_type` / `step_name`	NL2SQL process data
`complete`	Analysis completed
`error`	In-stream failure

Do not assume every event is identified through type, and do not deserialize every data object into one fixed schema. Inspect type, source, and step_type first. Preserve raw_data for an unknown event, then ignore it or route it to compatibility handling.

Cancel a run¶

The init event provides the request_id required for cancellation. Only the user who started the request can cancel it.

Python:

result = raw.cancel_analyze({"request_id": request_id})

Go:

result, err := client.CancelAnalyze(ctx, &sdk.CancelAnalyzeRequest{
    RequestID: requestID,
})

The cancellation response contains request_id, status, user_id, and user_name. Canceling a local context or closing the connection is not the same as requesting server-side cancellation; call the cancellation method when backend work must stop.

Failure layers¶

Stage	Typical failure	Handling
Before the stream opens	Authentication, permission, empty question, HTTP error	Do not enter the read loop; correct the request first
During reads	Network interruption, read timeout, malformed SSE	Close the stream and record the last event and request ID
In-stream `error`	Retrieval, SQL, or analysis step failure	Show the service error; do not mark received chunks as a complete answer
EOF before `complete`	Connection ends without an explicit completion event	Treat the run as incomplete and retain diagnostic data

A streamed analysis may already have done partial work. Before retrying automatically, verify that a retry is safe and create a new run record. Never concatenate chunks from two runs into one answer.

Sessions and persistence¶

Use session_id to continue a session and session_name to supply its name. An application-owned run record should retain at least:

session ID;
request ID from the init event;
question and a summary of the data-source configuration;
ordered events or the fields required for audit;
whether complete arrived, whether the run was canceled, and the final error.

Persisting provenance and completion state makes a run much easier to reproduce than storing rendered answer text alone.