minimax · 对话补全

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

def sse_lines(chunks):
    """增量解码 UTF-8，并跨网络分块正确识别 LF、CR 和 CRLF。"""
    import codecs
    decoder = codecs.getincrementaldecoder("utf-8")()
    line, skip_lf = [], False
    for chunk in chunks:
        for char in decoder.decode(chunk):
            if skip_lf:
                skip_lf = False
                if char == "\n":
                    continue
            if char in ("\r", "\n"):
                yield "".join(line)
                line = []
                skip_lf = char == "\r"
            else:
                line.append(char)
    decoder.decode(b"", final=True)  # 不接受 EOF 处截断的 UTF-8 字符。
    if line:
        yield "".join(line)


def iter_sse(lines):
    """只在空行处派发完整 SSE 帧。"""
    event, data_lines = "message", []
    first_line = True
    for line in lines:
        if first_line:
            line = line.removeprefix("\ufeff")
            first_line = False
        if line == "":
            if data_lines:
                yield event, "\n".join(data_lines)
            event, data_lines = "message", []
            continue
        if line.startswith(":"):
            continue  # 心跳或注释
        field, separator, value = line.partition(":")
        if separator and value.startswith(" "):
            value = value[1:]
        if field == "event":
            event = value or "message"
        elif field == "data":
            data_lines.append(value)
    # EOF 前没有空行的半帧不能当作完整事件；由下层检查终止事件。

def stream_text(events):
    completed = False
    for event, data in events:
        if data.strip() == "[DONE]":
            completed = True
            break
        item = json.loads(data)
        if event == "error" or item.get("error"):
            raise RuntimeError(item)
        for choice in item.get("choices", []):
            reason = choice.get("finish_reason")
            if reason in ("length", "content_filter"):
                raise RuntimeError("输出未完整完成：" + reason)
            delta = choice.get("delta", {})
            if delta.get("refusal"):
                raise RuntimeError(delta["refusal"])
            text = delta.get("content")
            if isinstance(text, str):
                yield text
        # usage 帧的 choices 可为空；finish_reason 后继续等 [DONE]。
    if not completed:
        raise RuntimeError("连接在 [DONE] 前结束，请保留已收内容并检查状态。")

def main():
    path = "/v1/chat/completions"
    body = {"model": MODEL_ID, "messages": [{"role": "user", "content": "用三句话介绍杭州。"}], "stream": True}
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"],
               "Accept": "text/event-stream"}
    headers.update({})
    with requests.post(BASE_URL + path, headers=headers, json=body,
                       stream=True, timeout=(10, 120)) as response:
        response.raise_for_status()
        if "text/event-stream" not in response.headers.get("Content-Type", ""):
            raise RuntimeError("预期 SSE，请检查返回的 Content-Type 和错误信息。")
        events = iter_sse(sse_lines(response.iter_content(chunk_size=128)))
        for text in stream_text(events):
            print(text, end="", flush=True)
    print()

if __name__ == "__main__":
    main()

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

PARAMETERS = {
    "type": "object",
    "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
    "required": ["a", "b"],
    "additionalProperties": False,
}


def run_tool(name, args):
    # 工具名和参数都先校验；只执行应用主动注册的函数。
    if name != "add_numbers" or not isinstance(args, dict) or set(args) != {"a", "b"}:
        raise ValueError("未知工具或参数不匹配")
    if any(type(args[key]) not in (int, float) for key in ("a", "b")):
        raise ValueError("a 和 b 必须为数字")
    return {"sum": args["a"] + args["b"]}

def main():
    tools = [{"type": "function", "function": {
        "name": "add_numbers", "description": "计算两个数字之和", "parameters": PARAMETERS}}]
    messages = [{"role": "user", "content": "请用工具计算 2 加 3。"}]
    first = request_json("/v1/chat/completions", {
        "model": MODEL_ID, "messages": messages, "tools": tools, "tool_choice": "auto"})
    assistant = first["choices"][0]["message"]
    calls = assistant.get("tool_calls", [])
    if not calls:
        print(assistant.get("content", ""))
        return
    messages.append(assistant)  # 保留整条消息、调用 ID 及模型返回的附加上下文。
    for call in calls:
        function = call["function"]
        result = run_tool(function["name"], json.loads(function["arguments"]))
        messages.append({"role": "tool", "tool_call_id": call["id"],
                         "content": json.dumps(result, ensure_ascii=False)})
    second = request_json("/v1/chat/completions", {
        "model": MODEL_ID, "messages": messages, "tools": tools, "tool_choice": "none"})
    print(second["choices"][0]["message"].get("content", ""))

if __name__ == "__main__":
    main()

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

def main():
    image = base64.b64encode(Path("input.png").read_bytes()).decode("ascii")
    result = request_json("/v1/chat/completions", {"model": MODEL_ID, "messages": [
        {"role": "user", "content": [
            {"type": "text", "text": "请描述这张图片。"},
            {"type": "image_url", "image_url": {"url": "data:image/png;base64," + image}}
        ]}]})
    print(result["choices"][0]["message"].get("content", ""))

if __name__ == "__main__":
    main()

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

SCHEMA = {
    "type": "object",
    "properties": {"city": {"type": "string"}, "days": {"type": "integer"}},
    "required": ["city", "days"],
    "additionalProperties": False,
}


def check_trip(value):
    # 业务侧仍校验结果，尤其要区分拒绝、截断和有效 JSON。
    if not isinstance(value, dict) or set(value) != {"city", "days"}:
        raise ValueError("返回对象不匹配 Schema")
    if not isinstance(value["city"], str) or type(value["days"]) is not int:
        raise ValueError("字段类型不匹配 Schema")
    return value

def main():
    result = request_json("/v1/chat/completions", {"model": MODEL_ID,
        "messages": [{"role": "user", "content": "提取旅行信息：杭州，3 天。"}],
        "response_format": {"type": "json_schema", "json_schema": {
            "name": "trip", "strict": True, "schema": SCHEMA}}})
    choice = result["choices"][0]
    message = choice["message"]
    if message.get("refusal") or choice.get("finish_reason") != "stop":
        raise RuntimeError("未得到完整结构化结果")
    print(check_trip(json.loads(message["content"])))

if __name__ == "__main__":
    main()