E-MapReduce:Call the EMR AI Assistant API

Supported regions

Complete client code

import json
import os
import sys

from alibabacloud_tea_openapi.client import Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models

# ENDPOINT: The endpoint for your instance's region. 
# For a complete list, see the table above.
ENDPOINT = "emrassistant.cn-hangzhou.aliyuncs.com"


def create_client() -> Client:
    return Client(open_api_models.Config(
        access_key_id=os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
        access_key_secret=os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
        endpoint=ENDPOINT,
    ))


_PARAMS = open_api_models.Params(
    action="ChatCompletion", version="2025-07-23", protocol="HTTPS",
    method="POST", pathname="/v1/chat/completions",
    auth_type="AK", style="ROA", req_body_type="json", body_type="json",
)


def chat(client: Client, question: str, session_id: str = None) -> str:
    """Sends a question to the assistant and streams the response. Returns a session ID for follow-up questions."""
    body = {"stream": True, "messages": [{"role": "user", "content": question}]}
    if session_id:
        body["sessionId"] = session_id

    request = open_api_models.OpenApiRequest(body=body)
    runtime = util_models.RuntimeOptions(connect_timeout=10_000, read_timeout=300_000)

    try:
        for resp in client.call_sseapi(_PARAMS, request, runtime):
            data = resp.event.data if resp.event else None
            if not data or data == "[DONE]":      # [DONE]: Marks the end of the response stream.
                continue
            msg = json.loads(data)
            if msg.get("type") == "run_started":
                session_id = msg.get("sessionId")        # Capture the session ID.
            elif msg.get("type") == "text":
                sys.stdout.write(msg.get("text", ""))    # Stream output in real time.
                sys.stdout.flush()
    except Exception as e:
        print(f"\n[API call failed] {e}", file=sys.stderr)
        print("Check that your AccessKey is correct, the EMR AI Assistant service is activated, "
              "and the endpoint matches your instance's region.", file=sys.stderr)
        return session_id
    print()
    return session_id

Slow SQL diagnosis

client = create_client()
chat(client,
     "My Serverless StarRocks instance c-xxxxxxxx (cn-hangzhou) has been slow recently. "
     "Help me diagnose the top slow SQL queries, identify bottlenecks, and provide optimization suggestions.")

The assistant automatically identifies the longest-running queries, locates SQL patterns that consume high resources, analyzes trends, and provides optimization suggestions for specific slow queries.

Health check

client = create_client()
chat(client,
     "Perform a health check on Serverless StarRocks instance c-xxxxxxxx (cn-hangzhou) and generate a report.")

The assistant performs a comprehensive check on the instance and outputs a health score, a list of anomalies, and improvement suggestions.

Slow SQL diagnosis output

The assistant automatically locates the instance, retrieves the queries that recently consumed the most resources, and provides a profile-level diagnosis:

Instance: c-xxxxxxxx (StarRocks 3.2.4, Running)
Most CPU-intensive query in the last hour: An information_schema metadata query
Diagnosis: Total time 102 ms / CPU 43.5 ms / Peak memory 3.21 MB / Wait ratio 80.88%
Bottleneck operators: SCHEMA_SCAN(node 7) 9.26 ms, SCHEMA_SCAN(node 0) 8.16 ms, HASH_JOIN(node 5) 1.28 MB
Conclusion: This query has a low workload and does not require optimization. If it occurs frequently, we recommend that you cache metadata on the client or use a connection pool.

Health check output

Instance: c-xxxxxxxx (StarRocks 3.2.4, onBareMetal / lakehouse, Running)
Health score: 100/100
Key metrics: Last 24h: Critical 0 / Warning 0 / Info 0. The overall stability is excellent.
Anomalies: None
Improvement suggestions: Evaluate a version upgrade, enable automatic minor version upgrades, set up regular health checks and alerts, and assess resource specifications based on business growth.

Specify the task and target

The assistant decides whether to call the diagnostic API based on your prompt. Include both an action and a target:

• Action: Use clear verbs, such as check, diagnose, identify, or optimize.

• Target: The instance ID (which starts with c-) and its region.

If you provide both, the assistant triggers a diagnosis. If either is missing, the request falls back to knowledge-based Q&A, and no instance data is retrieved.

Understand the response

The data collected and the results returned vary based on your request:

Interpret empty results correctly

If an instance has a low query volume, a slow SQL diagnosis may return "no significant slow queries found", and a health check may report "low workload and sufficient resources". This means the instance is healthy and does not indicate an API call failure.