Streaming API

MuAPI 支持 LLM 文本生成端点的 Server-Sent Events（SSE） 流式传输。生成 token 时会逐步发送结果，因此聊天 UI 和交互式应用可以实时显示内容。

Available Streaming Endpoints（可用的流式端点）

Endpoint	Model	Description
POST /api/v1/gemini-flash/stream	Gemini 3 Flash	支持视觉能力的快速多模态 LLM

How Streaming Works（流式传输原理）

向 /stream 端点发送 POST 请求，并附上请求 payload。
以流的形式读取响应，响应由 text/event-stream 事件组成。
每个事件都包含一个表示部分内容的 JSON delta。
流结束时会发送 data: [DONE] 标记。

计费： 流结束后根据实际 token 用量计算费用——Gemini 3 Flash 的输入 token 为每百万 $0.30，输出 token 为每百万 $1.80。钱包最低余额要求为 $1.00。

Request Format（请求格式）

Method： POST
Authentication： x-api-key header
Content-Type： application/json
Response Content-Type： text/event-stream

Payload

{
  "prompt": "Explain quantum entanglement in simple terms.",
  "image_url": "https://example.com/image.jpg",
  "system_prompt": "You are a concise science communicator."
}

Field	Type	Required	Description
prompt	string	是	用户消息或指令
image_url	string（URL）	否	多模态请求可选的图片
system_prompt	string	否	用于控制模型行为的系统级指令

SSE Response Format（SSE 响应格式）

每个事件行都以 data: 开头，后面跟一个 JSON 对象：

data: {"id":"chatcmpl-abc","choices":[{"index":0,"delta":{"content":"Quantum"},"finish_reason":null}]}

data: {"id":"chatcmpl-abc","choices":[{"index":0,"delta":{"content":" entanglement"},"finish_reason":null}]}

data: {"id":"chatcmpl-abc","choices":[{"index":0,"delta":{"content":"..."},"finish_reason":"stop"}],"usage":{"prompt_tokens":12,"completion_tokens":80,"total_tokens":92}}

data: [DONE]

Code Examples（代码示例）

Python（httpx — 推荐用于异步场景）

import httpx

API_KEY = "your_api_key_here"

with httpx.Client(timeout=120) as client:
    with client.stream(
        "POST",
        "https://api.muapi.ai/api/v1/gemini-flash/stream",
        headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
        json={
            "prompt": "Write a short poem about the ocean.",
            "system_prompt": "You are a creative poet."
        }
    ) as response:
        for line in response.iter_lines():
            if line.startswith("data: "):
                data = line[6:]
                if data == "[DONE]":
                    break
                import json
                chunk = json.loads(data)
                delta = chunk["choices"][0]["delta"].get("content", "")
                if delta:
                    print(delta, end="", flush=True)
        print()

Python（requests）

import requests
import json

API_KEY = "your_api_key_here"

response = requests.post(
    "https://api.muapi.ai/api/v1/gemini-flash/stream",
    headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
    json={
        "prompt": "Summarize the history of the internet.",
        "system_prompt": "Be concise and factual."
    },
    stream=True,
    timeout=120
)

for line in response.iter_lines():
    if line:
        line = line.decode("utf-8")
        if line.startswith("data: "):
            data = line[6:]
            if data == "[DONE]":
                break
            chunk = json.loads(data)
            delta = chunk["choices"][0]["delta"].get("content", "")
            if delta:
                print(delta, end="", flush=True)

print()

Python（with image — 多模态）

import httpx
import json

API_KEY = "your_api_key_here"

with httpx.Client(timeout=120) as client:
    with client.stream(
        "POST",
        "https://api.muapi.ai/api/v1/gemini-flash/stream",
        headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
        json={
            "prompt": "Describe what you see in this image.",
            "image_url": "https://example.com/photo.jpg"
        }
    ) as response:
        for line in response.iter_lines():
            if line.startswith("data: "):
                data = line[6:]
                if data == "[DONE]":
                    break
                chunk = json.loads(data)
                delta = chunk["choices"][0]["delta"].get("content", "")
                if delta:
                    print(delta, end="", flush=True)
        print()

JavaScript / Node.js（fetch）

const API_KEY = "your_api_key_here";

async function streamGeminiFlash(prompt, systemPrompt = null) {
  const body = { prompt };
  if (systemPrompt) body.system_prompt = systemPrompt;

  const response = await fetch("https://api.muapi.ai/api/v1/gemini-flash/stream", {
    method: "POST",
    headers: {
      "x-api-key": API_KEY,
      "Content-Type": "application/json",
    },
    body: JSON.stringify(body),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const chunk = decoder.decode(value, { stream: true });
    for (const line of chunk.split("\n")) {
      if (line.startsWith("data: ")) {
        const data = line.slice(6).trim();
        if (data === "[DONE]") return;
        try {
          const parsed = JSON.parse(data);
          const delta = parsed.choices?.[0]?.delta?.content ?? "";
          if (delta) process.stdout.write(delta);
        } catch {}
      }
    }
  }
}

streamGeminiFlash(
  "Explain how neural networks learn.",
  "You are a clear technical writer."
).then(() => console.log());

JavaScript（浏览器——通过 fetch 实现 EventSource 替代方案）

const API_KEY = "your_api_key_here";

async function streamToElement(prompt, targetElement) {
  const response = await fetch("https://api.muapi.ai/api/v1/gemini-flash/stream", {
    method: "POST",
    headers: {
      "x-api-key": API_KEY,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ prompt }),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const text = decoder.decode(value, { stream: true });
    for (const line of text.split("\n")) {
      if (line.startsWith("data: ")) {
        const data = line.slice(6).trim();
        if (data === "[DONE]") return;
        try {
          const parsed = JSON.parse(data);
          const delta = parsed.choices?.[0]?.delta?.content ?? "";
          if (delta) targetElement.textContent += delta;
        } catch {}
      }
    }
  }
}

// Usage
const outputDiv = document.getElementById("output");
streamToElement("Write a haiku about AI.", outputDiv);

cURL

curl -X POST "https://api.muapi.ai/api/v1/gemini-flash/stream" \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "What are the main causes of climate change?"}' \
  --no-buffer

如果只想查看文本内容（去掉 SSE 封装），可以使用：

curl -X POST "https://api.muapi.ai/api/v1/gemini-flash/stream" \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "List 5 programming best practices."}' \
  --no-buffer \
  | grep "^data: " \
  | grep -v "\[DONE\]" \
  | sed 's/^data: //' \
  | python3 -c "
import sys, json
for line in sys.stdin:
    try:
        chunk = json.loads(line)
        delta = chunk['choices'][0]['delta'].get('content', '')
        print(delta, end='', flush=True)
    except: pass
print()
"

TypeScript（带类型安全）

const API_BASE = "https://api.muapi.ai/api/v1";

interface StreamChunk {
  id: string;
  choices: Array<{
    index: number;
    delta: { content?: string; role?: string };
    finish_reason: string | null;
  }>;
  usage?: {
    prompt_tokens: number;
    completion_tokens: number;
    total_tokens: number;
  };
}

async function streamGeminiFlash(
  prompt: string,
  options: { imageUrl?: string; systemPrompt?: string } = {},
  apiKey: string
): Promise<string> {
  const body: Record<string, string> = { prompt };
  if (options.imageUrl) body.image_url = options.imageUrl;
  if (options.systemPrompt) body.system_prompt = options.systemPrompt;

  const response = await fetch(`${API_BASE}/gemini-flash/stream`, {
    method: "POST",
    headers: { "x-api-key": apiKey, "Content-Type": "application/json" },
    body: JSON.stringify(body),
  });

  if (!response.ok) throw new Error(`HTTP ${response.status}`);

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let fullText = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const text = decoder.decode(value, { stream: true });
    for (const line of text.split("\n")) {
      if (!line.startsWith("data: ")) continue;
      const data = line.slice(6).trim();
      if (data === "[DONE]") return fullText;
      try {
        const chunk: StreamChunk = JSON.parse(data);
        const delta = chunk.choices[0]?.delta?.content ?? "";
        fullText += delta;
        process.stdout.write(delta);
      } catch {}
    }
  }

  return fullText;
}

// Usage
streamGeminiFlash(
  "Describe the future of AI in healthcare.",
  { systemPrompt: "Be optimistic but realistic." },
  "your_api_key_here"
).then(() => console.log());

Comparison: Streaming vs Standard（流式与标准请求对比）

Feature	Standard（/gemini-flash）	Streaming（/gemini-flash/stream）
Response	request_id → 轮询结果	实时 SSE token 流
Latency to first token	较高（先完成整次生成）	较低（token 立即到达）
Best for	工作流、自动化、批处理	聊天 UI、实时显示
Webhook support	支持	不支持（响应本身就是流）
Billing	调用结束后按 token 计费	流结束后按 token 计费
Minimum balance	$1.00	$1.00

Error Handling（错误处理）

如果流式传输过程中发生错误，关闭连接前会发送一个错误事件：

data: {"error": "upstream provider timeout"}

客户端应始终处理这种情况：

chunk = json.loads(data)
if "error" in chunk:
    print(f"Stream error: {chunk['error']}")
    break
delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")

Pricing（价格）

Gemini 3 Flash 流式请求采用 token 计费，并在流结束后扣费：

Token Type	Rate
输入 token	每百万 $0.30
输出 token	每百万 $1.80

示例： 一个请求使用 500 个输入 token 和 800 个输出 token 时，费用为：

输入：500 × $0.30 / 1,000,000 = $0.00015
输出：800 × $1.80 / 1,000,000 = $0.00144
总计：约 $0.0016