AiMiaoBi API best practices

package org.example.writing;

import com.alibaba.fastjson2.JSONObject;
import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.service.aimiaobi20230801.AsyncClient;
import com.aliyun.sdk.service.aimiaobi20230801.models.*;
import darabonba.core.ResponseIterable;
import darabonba.core.client.ClientOverrideConfiguration;

import java.util.List;

public class RunWritingV2DirectWritingDemo {
    private static final String ALIBABA_CLOUD_ACCESS_KEY_ID = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String ALIBABA_CLOUD_ACCESS_KEY_SECRET = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    private static final String WORKSPACE_ID = System.getenv("WORKSPACE_ID");
    private static final String ENDPOINT = System.getenv().getOrDefault("DOMAIN", "aimiaobi.cn-beijing.aliyuncs.com");

    public static AsyncClient asyncClient() {
        return AsyncClient.builder().credentialsProvider(
                StaticCredentialProvider.create(Credential.builder().accessKeyId(ALIBABA_CLOUD_ACCESS_KEY_ID).accessKeySecret(ALIBABA_CLOUD_ACCESS_KEY_SECRET).build())
        ).serviceConfiguration(Configuration.create().setSignatureVersion(SignatureVersion.V3)).overrideConfiguration(
                ClientOverrideConfiguration.create().setProtocol("HTTPS").setEndpointOverride(ENDPOINT)
        ).build();
    }

    public static void main(String[] args) {
        directWriting();
    }

    public static void directWriting() {
        String prompt = "Write an article about machine learning";
        RunWritingV2Request.Builder builder = RunWritingV2Request.builder().workspaceId(WORKSPACE_ID);
        // Enable web search and source tracing.
        builder.sourceTraceMethod("modelSourceTrace").useSearch(true);
        builder.prompt(prompt);

        AsyncClient client = asyncClient();

        ResponseIterable<RunWritingV2ResponseBody> runWritingV2ResponseBodies = client.runWritingV2WithResponseIterable(builder.build());
        GenerateTraceability generateTraceability = null;
        for (RunWritingV2ResponseBody item : runWritingV2ResponseBodies) {
            System.out.println(JSONObject.toJSONString(item));
            if (item.getHeader().getErrorMessage() != null) {
                System.out.println("Writing failed: " + item.getHeader().getErrorMessage());
                break;
            }
            if ("task-progress-start-generating".equals(item.getHeader().getEvent())) {
                System.out.println(item.getPayload().getOutput().getText());
            }

            generateTraceability = item.getPayload().getOutput().getGenerateTraceability();
        }
        if (generateTraceability != null) {
            // List of articles actually referenced by the model.
            List<GenerateTraceability.News> news = generateTraceability.getNews();
            System.out.println(
                    JSONObject.toJSONString(news)
            );
        }
    }

}

import os
from alibabacloud_tea_openapi_sse.client import Client as OpenApiClient
from alibabacloud_tea_openapi_sse import models as open_api_models
from alibabacloud_tea_util_sse import models as open_api_util_models
import asyncio
import json

endpoint = 'aimiaobi.cn-beijing.aliyuncs.com'
# Hardcoding your AccessKey in code can lead to a credential leak, compromising the security of your account resources. The following code is for reference only.
# We recommend that you use a more secure method, such as STS. For more information about authentication methods, see: https://help.aliyun.com/en/sdk/developer-reference/v2-manage-python-access-credentials
access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')

access_key_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')

workspace_id = os.environ.get('WORKSPACE_ID')

# For API details, see: https://api.aliyun.com/api/AiMiaoBi/2023-08-01/RunWritingV2?spm=a2c4g.11186623.0.0.201c715eKrroz5&RegionId=cn-beijing

def _create_client() -> OpenApiClient:
    config = open_api_models.Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        endpoint=endpoint
    )
    return OpenApiClient(config)

def create_sse_api_info(action) -> open_api_models.Params:
    """
    Creates the parameter object for an API call.
    """
    params = open_api_models.Params(
        # API operation name
        action=action,
        # API version
        version='2023-08-01',
        # Protocol
        protocol='HTTPS',
        # HTTP method
        method='POST',
        auth_type='AK',
        style='RPC',
        pathname="",
        # Request body format
        req_body_type='formData',
        # Response body format
        body_type='sse'
    )
    return params

class SseClient:
    def __init__(self) -> None:
        # Hardcoding your AccessKey in code can lead to a credential leak, compromising the security of your account resources. The following code is for reference only.
        # We recommend that you use a more secure method, such as STS. For more information about authentication methods, see: https://help.aliyun.com/en/sdk/developer-reference/v2-manage-python-access-credentials
        self.endpoint = endpoint
        self._runtime = open_api_util_models.RuntimeOptions(read_timeout=1000 * 100)
        self._client = _create_client()

    async def run_writing(self, body):
        request = open_api_models.OpenApiRequest(body=body)
        return self._client.call_sse_api_async(params=create_sse_api_info("RunWritingV2"), request=request,
                                               runtime=self._runtime)

async def test_run_writing():
    client = SseClient()
    # Direct writing
    writing_params = {
        "Prompt": "Write an article about machine learning",
        "UseSearch": True,
        "SourceTraceMethod": "modelSourceTrace",
        "WorkspaceId": workspace_id
    }
    async for item in await  client.run_writing(writing_params):
        try:
            data = json.loads(item.get('event').data)
            print(json.dumps(data, ensure_ascii=False))
        except json.JSONDecodeError as e:
            print(e)
            print('------json.JSONDecodeError--------')

if __name__ == '__main__':
    asyncio.run(test_run_writing())

package org.example.writing;

import com.alibaba.fastjson2.JSONObject;
import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.service.aimiaobi20230801.AsyncClient;
import com.aliyun.sdk.service.aimiaobi20230801.models.GenerateTraceability;
import com.aliyun.sdk.service.aimiaobi20230801.models.RunWritingV2Request;
import com.aliyun.sdk.service.aimiaobi20230801.models.RunWritingV2ResponseBody;
import darabonba.core.ResponseIterable;
import darabonba.core.client.ClientOverrideConfiguration;
import lombok.Data;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RunWritingV2WithTemplate {
    private static final String ALIBABA_CLOUD_ACCESS_KEY_ID = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String ALIBABA_CLOUD_ACCESS_KEY_SECRET = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    private static final String WORKSPACE_ID = System.getenv("WORKSPACE_ID");
    private static final String ENDPOINT = System.getenv().getOrDefault("DOMAIN", "aimiaobi.cn-beijing.aliyuncs.com");

    public static AsyncClient asyncClient() {
        return AsyncClient.builder().credentialsProvider(
                StaticCredentialProvider.create(Credential.builder().accessKeyId(ALIBABA_CLOUD_ACCESS_KEY_ID).accessKeySecret(ALIBABA_CLOUD_ACCESS_KEY_SECRET).build())
        ).serviceConfiguration(Configuration.create().setSignatureVersion(SignatureVersion.V3)).overrideConfiguration(
                ClientOverrideConfiguration.create().setProtocol("HTTPS").setEndpointOverride(ENDPOINT)
        ).build();
    }

    public static void main(String[] args) {
        WritingResult result = templateWriting();
        // Final consolidated output (aligned with RunWritingV2Outline.java / RunWritingV2WithTemplate.py)
        System.out.println("==================== [Writing Result] ====================");
        System.out.println(JSONObject.toJSONString(result));
    }

    /**
     * Template-based writing: Media - News Report.
     *
     * <p>Use {@code promptMode="Template"} and pass the following fields in {@code writingParams}:</p>
     * <ul>
     *   <li>{@code topic}           : The topic</li>
     *   <li>{@code corePerspective} : The core perspective</li>
     *   <li>{@code newsStructure}   : The news structure (for example, "Chronological structure")</li>
     * </ul>
     *
     * <p>This example also enables web search and model source tracing ({@code modelSourceTrace}).</p>
     *
     * @return The results collected during the writing process ({@link WritingResult}), with a structure aligned with
     *         {@link RunWritingV2Outline.WritingResult}. (Template-based writing does not involve an outline, so `outlines` is null).
     */
    public static WritingResult templateWriting() {
        AsyncClient client = asyncClient();

        // For more writing style templates, see the API documentation: https://help.aliyun.com/en/model-studio/api-aimiaobi-2023-08-01-listwritingstyles
        String writingScene = "media";
        String writingStyle = "新闻报道";

        // Template fields (writingParams)
        Map<String, String> writingParams = new HashMap<>();
        writingParams.put("topic", "Chang’e-6 achieves the world’s first sample return from the far side of the Moon");
        writingParams.put("corePerspective", "On June 25, 2024, the Chang’e-6 return capsule landed successfully in the designated area of Siziwang Banner, Inner Mongolia, "
                + "bringing back about 1,935.3 g of samples from the far side of the Moon and achieving the first far-side lunar sample return in human history. "
                + "The 53-day mission completed key milestones, including launch and orbit insertion, Earth-Moon transfer, lunar braking, circumlunar flight, landing and sampling, "
                + "takeoff from the far side, rendezvous and docking in lunar orbit, trans-Earth injection, and reentry and recovery. This milestone achievement "
                + "demonstrates China’s strength in space technology independence and self-reliance, contributes Chinese expertise and solutions to international cooperation "
                + "in lunar and deep-space exploration, and boosts the confidence and determination of the nation to reach new scientific and technological heights.");
        writingParams.put("newsStructure", "编年体结构");

        RunWritingV2Request.Builder builder = RunWritingV2Request.builder()
                .workspaceId(WORKSPACE_ID)
                .step("Writing")
                .promptMode("Template")
                .writingScene(writingScene)
                .writingStyle(writingStyle)
                .writingParams(writingParams)
                .useSearch(true)
                .sourceTraceMethod("modelSourceTrace");

        // A process result collector aligned with WritingResult in RunWritingV2Outline.java.
        // (Template-based writing does not involve an outline, so the `outlines` field is null.)
        WritingResult result = new WritingResult();

        ResponseIterable<RunWritingV2ResponseBody> stream = client.runWritingV2WithResponseIterable(builder.build());
        for (RunWritingV2ResponseBody item : stream) {
            // Print the raw event body
            System.out.println(JSONObject.toJSONString(item));

            // Task failed: print the error and stop.
            if (item.getHeader().getErrorMessage() != null) {
                System.err.println("ErrorCode: " + item.getHeader().getErrorCode()
                        + ", ErrorMessage: " + item.getHeader().getErrorMessage());
                break;
            }

            RunWritingV2ResponseBody.Output output = item.getPayload().getOutput();
            if (output == null) {
                continue;
            }

            // Incremental content: The text is continuously updated with each streaming event. The final version is the last one received.
            if (output.getText() != null && !output.getText().isEmpty()) {
                result.setText(output.getText());
            }

            // Source tracing: Can be included at any stage. Refresh when received.
            GenerateTraceability traceability = output.getGenerateTraceability();
            if (traceability != null && traceability.getNews() != null) {
                result.setNews(traceability.getNews());
            }
        }
        return result;
    }

    // ============================== Data Structures ==============================

    /**
     * Intermediate result for template-based writing.
     *
     * <p>The structure is aligned with {@link RunWritingV2Outline.WritingResult} for consistent upstream consumption.
     * Template-based writing does not involve an outline, so it does not include the `outlines` field.</p>
     */
    @Data
    public static class WritingResult {
        /**
         * The final generated full text.
         * <p>This is continuously updated by streaming events; the final version is the last one received.</p>
         */
        private String text;

        /**
         * Source tracing information: A list of articles referenced in the text.
         * <p>This is non-null only when source tracing (for example, {@code sourceTraceMethod="modelSourceTrace"}) is enabled and there are matches.</p>
         * <p>The articles correspond one-to-one with the {@code [[n]]} markers in the text (index starts at 1).</p>
         */
        private List<GenerateTraceability.News> news;
    }

}

"""
Template-based writing example (Python).

Scenario: Media - News Report
  Use `promptMode="Template"` and pass template fields such as topic, core perspective, and news structure via `writingParams`.
  This enables web search and model source tracing (`modelSourceTrace`).

API Reference:
  https://api.aliyun.com/api/AiMiaoBi/2023-08-01/RunWritingV2
"""
import asyncio
import json
import os
from typing import Any, Dict

from alibabacloud_tea_openapi_sse import models as open_api_models
from alibabacloud_tea_openapi_sse.client import Client as OpenApiClient
from alibabacloud_tea_util_sse import models as open_api_util_models

# ============================== Environment Configuration ==============================
endpoint = os.environ.get("DOMAIN", "aimiaobi.cn-beijing.aliyuncs.com")
# Hardcoding your AccessKey in code can lead to a credential leak, compromising the security of your account resources. The following code is for reference only.
# We recommend that you use a more secure method, such as STS. For more information about authentication methods, see: https://help.aliyun.com/en/sdk/developer-reference/v2-manage-python-access-credentials
access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID")
access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
workspace_id = os.environ.get("WORKSPACE_ID")

# ============================== Client ==============================
def _create_client() -> OpenApiClient:
    config = open_api_models.Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        endpoint=endpoint,
    )
    return OpenApiClient(config)

def _create_sse_api_info(action: str) -> open_api_models.Params:
    return open_api_models.Params(
        action=action,
        version="2023-08-01",
        protocol="HTTPS",
        method="POST",
        auth_type="AK",
        style="RPC",
        pathname="",
        req_body_type="formData",
        body_type="sse",
    )

class SseClient:
    def __init__(self) -> None:
        self.endpoint = endpoint
        self._runtime = open_api_util_models.RuntimeOptions(read_timeout=1000 * 100)
        self._client = _create_client()

    async def run_writing(self, body: Dict[str, Any]):
        request = open_api_models.OpenApiRequest(body=body)
        return self._client.call_sse_api_async(
            params=_create_sse_api_info("RunWritingV2"),
            request=request,
            runtime=self._runtime,
        )

# ============================== Template-based Writing ==============================
async def template_writing() -> Dict[str, Any]:
    """
    Template-based writing: Media - News Report.

    Use `PromptMode="Template"` and pass the following in `WritingParams`:
      - topic           : The topic
      - corePerspective : The core perspective
      - newsStructure   : The news structure (for example, "Chronological structure")

    :return: A dictionary of results collected during writing, aligned with RunWritingV2OutlineWriting.py:
        {
          "text": "...",   # The final generated text (continuously updated by streaming events; the last one received is final).
          "news": [...]     # The list of referenced articles from source tracing.
        }
    """
    client = SseClient()
    # For more writing style templates, see the API documentation: https://help.aliyun.com/en/model-studio/api-aimiaobi-2023-08-01-listwritingstyles
    writing_scene = "media"
    writing_style = "新闻报道"

    # Template fields (writingParams)
    writing_params: Dict[str, str] = {
        "topic": "Chang’e-6 achieves the world’s first sample return from the far side of the Moon",
        "corePerspective": (
            "On June 25, 2024, the Chang’e-6 return capsule landed successfully in the designated area of Siziwang Banner, Inner Mongolia, "
            "bringing back about 1,935.3 g of samples from the far side of the Moon and achieving the first far-side lunar sample return in human history. "
            "The 53-day mission completed key milestones, including launch and orbit insertion, Earth-Moon transfer, lunar braking, circumlunar flight, landing and sampling, "
            "takeoff from the far side, rendezvous and docking in lunar orbit, trans-Earth injection, and reentry and recovery. This milestone achievement "
            "demonstrates China’s strength in space technology independence and self-reliance, contributes Chinese expertise and solutions to international cooperation "
            "in lunar and deep-space exploration, and boosts the confidence and determination of the nation to reach new scientific and technological heights."
        ),
        "newsStructure": "编年体结构",
    }

    body: Dict[str, Any] = {
        "WorkspaceId": workspace_id,
        "Step": "Writing",
        "PromptMode": "Template",
        "WritingScene": writing_scene,
        "WritingStyle": writing_style,
        "WritingParams": json.dumps(writing_params, ensure_ascii=False),
        # Enable web search
        "UseSearch": True,
        # Source tracing method: `modelSourceTrace` instructs the model to insert [[n]] reference markers at the end of text segments, with the index starting at 1.
        "SourceTraceMethod": "modelSourceTrace",
    }

    # A process result collector aligned with WritingResult in RunWritingV2OutlineWriting.py.
    # (Template-based writing does not involve an outline, so it does not include an `outlines` field.)
    result: Dict[str, Any] = {"text": None, "news": None}

    async for item in await client.run_writing(body):
        try:
            data = json.loads(item.get("event").data)
        except (json.JSONDecodeError, AttributeError) as e:
            print(f"[Parsing failed] {e}")
            continue

        # Print the raw event body (to align with Java's JSONObject.toJSONString(item)).
        print(json.dumps(data, ensure_ascii=False))

        header = data.get("Header", {}) or {}
        if header.get("ErrorMessage"):
            print(f"Writing failed: {header.get('ErrorMessage')}")
            break

        output = (data.get("Payload", {}) or {}).get("Output", {}) or {}

        # Incremental content: The text is continuously updated with each streaming event. The final version is the last one received.
        text = output.get("Text")
        if text:
            result["text"] = text

        # Source tracing: Can be included at any stage. Refresh when received.
        traceability = output.get("GenerateTraceability")
        if traceability and traceability.get("News"):
            result["news"] = traceability.get("News")

    # Final consolidated output
    print("==================== [Writing Result] ====================")
    print(json.dumps(result, ensure_ascii=False))
    return result

if __name__ == "__main__":
    asyncio.run(template_writing())

package org.example.writing;

import com.alibaba.fastjson2.JSONObject;
import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.service.aimiaobi20230801.AsyncClient;
import com.aliyun.sdk.service.aimiaobi20230801.models.*;
import darabonba.core.ResponseIterable;
import darabonba.core.client.ClientOverrideConfiguration;
import lombok.Data;

import java.util.List;
import java.util.stream.Collectors;

/**
 * A simplified example of step-by-step, outline-based writing.
 *
 * <p>Core two-step process:</p>
 * <ol>
 *   <li><b>OutlineGenerate</b>: Generate a tree-structured outline based on the prompt.</li>
 *   <li><b>Writing</b>: Searches for information for each section of the outline before writing the main text.</li>
 * </ol>
 */
public class RunWritingV2Outline {

    // ============================== Environment Configuration ==============================
    private static final String ALIBABA_CLOUD_ACCESS_KEY_ID = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String ALIBABA_CLOUD_ACCESS_KEY_SECRET = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    private static final String WORKSPACE_ID = System.getenv("WORKSPACE_ID");
    private static final String ENDPOINT = System.getenv().getOrDefault("DOMAIN", "aimiaobi.cn-beijing.aliyuncs.com");

    // ============================== Step and Event Constants ==============================
    private static final String STEP_OUTLINE_GENERATE = "OutlineGenerate";
    private static final String STEP_WRITING = "Writing";

    private static final String EVT_TASK_FAILED = "task-failed";
    private static final String EVT_OUTLINE_GENERATE = "task-outline-generate";
    private static final String EVT_OUTLINE_GENERATE_REASONING = "task-outline-generate-reasoning";
    private static final String EVT_OUTLINE_GENERATE_END = "task-outline-generate-end";
    private static final String EVT_OUTLINE_SEARCH_START = "task-outline-search-start";
    private static final String EVT_OUTLINE_SEARCH = "task-outline-search";

    // Fixed values for outline-based writing
    private static final String WRITING_STYLE_OUTLINE = "outlineWriting";
    private static final String WRITING_SCENE_OTHERS = "others";

    // ============================== Client ==============================
    public static AsyncClient asyncClient() {
        return AsyncClient.builder()
                .credentialsProvider(StaticCredentialProvider.create(
                        Credential.builder()
                                .accessKeyId(ALIBABA_CLOUD_ACCESS_KEY_ID)
                                .accessKeySecret(ALIBABA_CLOUD_ACCESS_KEY_SECRET)
                                .build()))
                .serviceConfiguration(Configuration.create().setSignatureVersion(SignatureVersion.V3))
                .overrideConfiguration(ClientOverrideConfiguration.create()
                        .setProtocol("HTTPS")
                        .setEndpointOverride(ENDPOINT))
                .build();
    }

    // ============================== Entry Point: Full Pipeline ==============================
    public static void main(String[] args) {
        AsyncClient client = asyncClient();
        String prompt = "Write an article about new energy vehicles, around 3000 words";

        // Step 1: Generate an outline
        System.out.println("==================== [Step 1] Generate Outline ====================");
        WritingResult outlineResult =  runWriting(client, prompt, STEP_OUTLINE_GENERATE, null);

        if (outlineResult.getOutlines() == null || outlineResult.getOutlines().isEmpty()) {
            System.err.println("Failed to get outline. Skipping text generation.");
            return;
        }
        System.out.println("Outline preview:");
        for (WritingOutline outline : outlineResult.getOutlines()) {
            System.out.print(outlineToTreeStr(outline, 0));
        }

        // Step 2: Write the main text based on the outline
        System.out.println("==================== [Step 2] Write Text ====================");
        WritingResult writingResult =  runWriting(client, prompt, STEP_WRITING, outlineResult.getOutlines());
        System.out.println(JSONObject.toJSONString(writingResult));
    }

    // ============================== Core Call ==============================

    private static WritingResult runWriting(AsyncClient client,
                                            String prompt,
                                            String step,
                                            List<WritingOutline> outlines) {
        RunWritingV2Request request = buildRequest(prompt, step, outlines);
        ResponseIterable<RunWritingV2ResponseBody> stream = client.runWritingV2WithResponseIterable(request);

        WritingResult result = new WritingResult();
        for (RunWritingV2ResponseBody item : stream) {

            System.out.println(JSONObject.toJSONString(item));

            String event = item.getHeader().getEvent();
            RunWritingV2ResponseBody.Output output = item.getPayload().getOutput();

            // Task failed: print the error and stop.
            if (EVT_TASK_FAILED.equals(event)) {
                System.err.println("ErrorCode: " + item.getHeader().getErrorCode()
                        + ", ErrorMessage: " + item.getHeader().getErrorMessage());
                break;
            }
            // Outline event: refresh the outline.
            else if (isOutlineGenerateEvent(event)) {
                if (EVT_OUTLINE_GENERATE_REASONING.equals(event)) {
                    System.out.println("[Outline Reasoning] " + output.getText());
                } else {
                    result.setOutlines(output.getOutlines());
                }
            }
            // Outline search event
            else if (isOutlineSearchEvent(event)) {
                printOutlineSearch(event, output);
            }
            // Body text event
            else {
                System.out.println("[Text Segment] " + JSONObject.toJSONString(output.getText()));
                result.setText(output.getText());
            }
            if ( output.getGenerateTraceability()!=null ){
                // Source tracing
                result.setNews(output.getGenerateTraceability().getNews());
            }
        }
        return result;
    }

    private static RunWritingV2Request buildRequest(String prompt, String step,
                                                    List<WritingOutline> outlines) {
        RunWritingV2Request.Builder builder = RunWritingV2Request.builder()
                .workspaceId(WORKSPACE_ID)
                // Outline-based writing always uses `outlineWriting` and `others`.
                .writingStyle(WRITING_STYLE_OUTLINE)
                .writingScene(WRITING_SCENE_OTHERS)
                .sourceTraceMethod("modelSourceTrace")
                // Enable web search
                .useSearch(true)
                .step(step);

        if (prompt != null) {
            builder.prompt(prompt);
        }
        if (outlines != null) {
            builder.outlineList(outlines);
        }
        return builder.build();
    }

    // ============================== Event Helpers ==============================

    private static boolean isOutlineGenerateEvent(String event) {
        return EVT_OUTLINE_GENERATE.equals(event)
                || EVT_OUTLINE_GENERATE_REASONING.equals(event)
                || EVT_OUTLINE_GENERATE_END.equals(event);
    }

    private static boolean isOutlineSearchEvent(String event) {
        return EVT_OUTLINE_SEARCH_START.equals(event) || EVT_OUTLINE_SEARCH.equals(event);
    }

    /** Prints simple start and hit information for the outline search. */
    private static void printOutlineSearch(String event, RunWritingV2ResponseBody.Output output) {
        if (EVT_OUTLINE_SEARCH_START.equals(event)) {
            System.out.println("[Outline Search Start] " + output.getText());
            return;
        }
        OutlineSearchResult sr = output.getSearchResult();
        if (sr == null || sr.getArticles() == null) {
            return;
        }
        System.out.println("[Outline Search Hit] " + sr.getOutline()
                + " (Query=" + sr.getQuery() + ", " + sr.getArticles().size() + " total): "
                + sr.getArticles().stream().map(OutlineWritingArticle::getTitle)
                        .collect(Collectors.joining(", ")));
    }

    // ============================== Utility Methods ==============================

    /** Converts a single outline into a tree-like text structure for console display. */
    public static String outlineToTreeStr(WritingOutline outline, int level) {
        StringBuilder sb = new StringBuilder();
        StringBuilder indent = new StringBuilder();
        for (int i = 0; i < level; i++) {
            indent.append("  ");
        }

        if (outline.getOutline() != null && !outline.getOutline().trim().isEmpty()) {
            String symbol = (level == 0) ? "┌─ " : "├─ ";
            sb.append(indent).append(symbol).append(outline.getOutline()).append("\n");
        }
        if (outline.getChildren() != null) {
            for (WritingOutline child : outline.getChildren()) {
                sb.append(outlineToTreeStr(child, level + 1));
            }
        }
        return sb.toString();
    }

    // ============================== Data Structures ==============================

    /** Intermediate result for outline-based writing. */
    @Data
    public static class WritingResult {
        /**
         * The tree structure of the outline.
         * <p>This is refreshed only during the {@code OutlineGenerate} stage and is {@code null} during the {@code Writing} stage.</p>
         * <p>This can be passed as the {@code outlineList} parameter for the next "Writing" step.</p>
         */
        private List<WritingOutline> outlines;

        /**
         * The final generated full text.
         * <p>This is assigned only during the {@code Writing} stage. It is continuously updated by streaming events; the final version is the last one received.</p>
         */
        private String text;

        /**
         * Source tracing information: A list of articles referenced in the text.
         * <p>This is non-null only when source tracing (for example, {@code sourceTraceMethod="modelSourceTrace"}) is enabled and there are matches.</p>
         * <p>The articles correspond one-to-one with the {@code [[n]]} markers in the text (index starts at 1).</p>
         */
        private java.util.List<GenerateTraceability.News> news;
    }
}

"""
Step-by-step outline-based writing example (Python).

Core two-step process:
  1) OutlineGenerate: Generate a tree-structured outline from the prompt.
  2) Writing: Based on the generated outline, write the main text with per-section search.

API Reference:
  https://api.aliyun.com/api/AiMiaoBi/2023-08-01/RunWritingV2
"""
import asyncio
import json
import os
from typing import Any, Dict, List, Optional

from alibabacloud_tea_openapi_sse import models as open_api_models
from alibabacloud_tea_openapi_sse.client import Client as OpenApiClient
from alibabacloud_tea_util_sse import models as open_api_util_models

# ============================== Environment Configuration ==============================
endpoint = os.environ.get("DOMAIN", "aimiaobi.cn-beijing.aliyuncs.com")
# Hardcoding your AccessKey in code can lead to a credential leak, compromising the security of your account resources. The following code is for reference only.
# We recommend that you use a more secure method, such as STS. For more information about authentication methods, see: https://help.aliyun.com/en/sdk/developer-reference/v2-manage-python-access-credentials
access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID")
access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
workspace_id = os.environ.get("WORKSPACE_ID")

# ============================== Step and Event Constants ==============================
STEP_OUTLINE_GENERATE = "OutlineGenerate"
STEP_WRITING = "Writing"

EVT_TASK_FAILED = "task-failed"
EVT_OUTLINE_GENERATE = "task-outline-generate"
EVT_OUTLINE_GENERATE_REASONING = "task-outline-generate-reasoning"
EVT_OUTLINE_GENERATE_END = "task-outline-generate-end"
EVT_OUTLINE_SEARCH_START = "task-outline-search-start"
EVT_OUTLINE_SEARCH = "task-outline-search"

# Fixed values for outline-based writing
WRITING_STYLE_OUTLINE = "outlineWriting"
WRITING_SCENE_OTHERS = "others"

# ============================== Client ==============================
def _create_client() -> OpenApiClient:
    config = open_api_models.Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        endpoint=endpoint,
    )
    return OpenApiClient(config)

def _create_sse_api_info(action: str) -> open_api_models.Params:
    return open_api_models.Params(
        action=action,
        version="2023-08-01",
        protocol="HTTPS",
        method="POST",
        auth_type="AK",
        style="RPC",
        pathname="",
        req_body_type="formData",
        body_type="sse",
    )

class SseClient:
    def __init__(self) -> None:
        self.endpoint = endpoint
        self._runtime = open_api_util_models.RuntimeOptions(read_timeout=1000 * 100)
        self._client = _create_client()

    async def run_writing(self, body: Dict[str, Any]):
        request = open_api_models.OpenApiRequest(body=body)
        return self._client.call_sse_api_async(
            params=_create_sse_api_info("RunWritingV2"),
            request=request,
            runtime=self._runtime,
        )

# ============================== Utility Methods ==============================
def outline_to_tree_str(outline: Dict[str, Any], level: int = 0) -> str:
    """Converts a single outline into a tree-like text structure for console display."""
    indent = "  " * level
    text = ""
    name = (outline.get("Outline") or outline.get("outline") or "").strip()
    if name:
        symbol = "┌─ " if level == 0 else "├─ "
        text += f"{indent}{symbol}{name}\n"
    children = outline.get("Children") or outline.get("children") or []
    for child in children:
        text += outline_to_tree_str(child, level + 1)
    return text

def _is_outline_generate_event(event: str) -> bool:
    return event in (
        EVT_OUTLINE_GENERATE,
        EVT_OUTLINE_GENERATE_REASONING,
        EVT_OUTLINE_GENERATE_END,
    )

def _is_outline_search_event(event: str) -> bool:
    return event in (EVT_OUTLINE_SEARCH_START, EVT_OUTLINE_SEARCH)

def _print_outline_search(event: str, output: Dict[str, Any]) -> None:
    if event == EVT_OUTLINE_SEARCH_START:
        print(f"[Outline Search Start] {output.get('Text', '')}")
        return
    sr = output.get("SearchResult")
    if not sr:
        return
    articles = sr.get("Articles") or []
    titles = ", ".join(a.get("Title", "") for a in articles)
    print(
        f"[Outline Search Hit] {sr.get('Outline', '')} "
        f"(Query={sr.get('Query', '')}, {len(articles)} total): {titles}"
    )

# ============================== Core Call ==============================
def _build_body(
        prompt: Optional[str],
        step: str,
        outlines: Optional[List[Dict[str, Any]]],
) -> Dict[str, Any]:
    body: Dict[str, Any] = {
        "WorkspaceId": workspace_id,
        # Outline-based writing always uses `outlineWriting` and `others`.
        "WritingStyle": WRITING_STYLE_OUTLINE,
        "WritingScene": WRITING_SCENE_OTHERS,
        # Enable web search
        "UseSearch": True,
        # Source tracing method: `modelSourceTrace` instructs the model to insert [[n]] reference markers at the end of text segments, with the index starting at 1.
        "SourceTraceMethod": "modelSourceTrace",
        "Step": step,
    }
    if prompt is not None:
        body["Prompt"] = prompt
    if outlines is not None:
        body["OutlineList"] = outlines
    return body

async def run_writing(
        client: SseClient,
        prompt: Optional[str],
        step: str,
        outlines: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """Initiates a RunWritingV2 call and processes events by type.

    Return format:
      {
        "outlines": [...],   # Refreshed during the `OutlineGenerate` stage
        "text": "...",        # Final text from the `Writing` stage
        "news": [...]         # Referenced articles from source tracing
      }
    """
    body = _build_body(prompt, step, outlines)
    result: Dict[str, Any] = {"outlines": None, "text": None, "news": None}

    async for item in await client.run_writing(body):
        try:
            data = json.loads(item.get("event").data)
        except (json.JSONDecodeError, AttributeError) as e:
            print(f"[Parsing failed] {e}")
            continue

        # Print the raw event body (to align with Java's JSONObject.toJSONString(item)).
        print(json.dumps(data, ensure_ascii=False))

        header = data.get("Header", {}) or {}
        payload = data.get("Payload", {}) or {}
        output = payload.get("Output", {}) or {}
        event = header.get("Event")

        # Task failed: print the error and stop.
        if event == EVT_TASK_FAILED:
            print(
                f"[Task Failed] ErrorCode={header.get('ErrorCode')}, "
                f"ErrorMessage={header.get('ErrorMessage')}"
            )
            break

        # Outline event: refresh the outline.
        if _is_outline_generate_event(event):
            if event == EVT_OUTLINE_GENERATE_REASONING:
                print(f"[Outline Reasoning] {output.get('Text', '')}")
            else:
                result["outlines"] = output.get("Outlines")

        # Outline search event
        elif _is_outline_search_event(event):
            _print_outline_search(event, output)

        # Body text event
        else:
            text = output.get("Text", "")
            if text:
                print(f"[Text Segment] {json.dumps(text, ensure_ascii=False)}")
                result["text"] = text

        # Source tracing information
        traceability = output.get("GenerateTraceability")
        if traceability and traceability.get("News"):
            result["news"] = traceability.get("News")

    return result

# ============================== Full Pipeline Example ==============================
async def test_outline_writing():
    client = SseClient()
    prompt = "Write an article about new energy vehicles, around 3000 words"

    # Step 1: Generate an outline
    print("==================== [Step 1] Generate Outline ====================")
    outline_result = await run_writing(client, prompt, STEP_OUTLINE_GENERATE)
    outlines = outline_result.get("outlines")
    if not outlines:
        print("Failed to get outline. Skipping text generation.")
        return

    print("Outline preview:")
    for outline in outlines:
        print(outline_to_tree_str(outline, 0), end="")

    # Step 2: Write the main text based on the outline
    print("==================== [Step 2] Write Text ====================")
    writing_result = await run_writing(client, prompt, STEP_WRITING, outlines=outlines)
    print(json.dumps(writing_result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    asyncio.run(test_outline_writing())

package com.aliyun.sdk.service.aimiaobi20230801;

import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.service.aimiaobi20230801.models.RunSummaryGenerateRequest;
import com.aliyun.sdk.service.aimiaobi20230801.models.RunSummaryGenerateResponseBody;
import com.aliyun.sdk.service.aimiaobi20230801.models.RunWritingRequest;
import com.aliyun.sdk.service.aimiaobi20230801.models.RunWritingResponseBody;
import com.google.gson.Gson;
import com.aliyun.sdk.service.aimiaobi20230801.AsyncClient;
import darabonba.core.ResponseIterable;
import darabonba.core.ResponseIterator;
import darabonba.core.client.ClientOverrideConfiguration;

import java.util.ArrayList;
import java.util.List;

/**
 * This class demonstrates how to use the summary generation API.
 */
public class RunSummaryGenerateTest {

    public static void main(String[] args) {
        StaticCredentialProvider provider = StaticCredentialProvider.create(Credential.builder()
                .accessKeyId(Constant.accessKeyId)
                .accessKeySecret(Constant.accessKeySecret)
                .build());

        AsyncClient client = AsyncClient.builder()
                .region("cn-beijing")
                .credentialsProvider(provider)
                .serviceConfiguration(Configuration.create().setSignatureVersion(SignatureVersion.V3)).overrideConfiguration(ClientOverrideConfiguration.create().setProtocol("HTTPS").setEndpointOverride("aimiaobi.cn-beijing.aliyuncs.com"))
                .build();

        RunSummaryGenerateRequest request = RunSummaryGenerateRequest.builder()
                .workspaceId(Constant.workspaceId)
                .prompt("Generate a summary of the following content in 100 to 200 words.")
                .content("Elastic Compute Service (ECS) is a high-performance, stable, reliable, and scalable IaaS (Infrastructure as a Service) cloud computing service provided by Alibaba Cloud. ECS eliminates the need for upfront IT hardware procurement, allowing you to use servers as conveniently and efficiently as public utilities like water, electricity, and natural gas. This enables on-demand provisioning and elastic scaling of compute resources. Alibaba Cloud ECS continuously offers innovative server instances to meet diverse business needs and help your business grow.")
                .build();

        ResponseIterable<RunSummaryGenerateResponseBody> x = client.runSummaryGenerateWithResponseIterable(request);

        ResponseIterator<RunSummaryGenerateResponseBody> iterator = x.iterator();
        while (iterator.hasNext()) {
            System.out.println("----event----");
            RunSummaryGenerateResponseBody event = iterator.next();
            System.out.println(new Gson().toJson(event));
        }

        System.out.println("ALL***********************");
        System.out.println("Headers for a successful request:");
        System.out.println(x.getStatusCode());
        System.out.println(x.getHeaders());
    }
}