E-MapReduce:Call the EMR AI Assistant API

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

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

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

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.

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.