OpenAI · Responses

# 安装依赖：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]":
            break
        item = json.loads(data)
        kind = item.get("type", event)
        if kind in ("error", "response.failed", "response.incomplete") or item.get("error"):
            raise RuntimeError(item)
        if kind == "response.refusal.delta":
            raise RuntimeError(item.get("delta", "请求未完成"))
        if kind == "response.output_text.delta":
            yield item.get("delta", "")
        elif kind == "response.completed":
            status = item.get("response", {}).get("status")
            if status != "completed":
                raise RuntimeError(item)
            completed = True
            break
    if not completed:
        raise RuntimeError("连接在 response.completed 前结束。")

def main():
    path = "/v1/responses"
    body = {"model": MODEL_ID, "input": "用三句话介绍杭州。", "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", "name": "add_numbers",
              "description": "计算两个数字之和", "parameters": PARAMETERS}]
    items = [{"role": "user", "content": "请用工具计算 2 加 3。"}]
    first = request_json("/v1/responses", {
        "model": MODEL_ID, "input": items, "tools": tools, "tool_choice": "auto"})
    output = first.get("output", [])
    calls = [item for item in output if item.get("type") == "function_call"]
    if not calls:
        print(first)
        return
    items.extend(output)  # 一并保留 reasoning 等输出项，不只保留 function_call。
    for call in calls:
        result = run_tool(call["name"], json.loads(call["arguments"]))
        items.append({"type": "function_call_output", "call_id": call["call_id"],
                      "output": json.dumps(result, ensure_ascii=False)})
    second = request_json("/v1/responses", {
        "model": MODEL_ID, "input": items, "tools": tools, "tool_choice": "none"})
    for item in second.get("output", []):
        for part in item.get("content", []):
            if part.get("type") == "output_text":
                print(part["text"])

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/responses", {"model": MODEL_ID, "input": [
        {"role": "user", "content": [
            {"type": "input_text", "text": "请描述这张图片。"},
            {"type": "input_image", "image_url": "data:image/png;base64," + image}
        ]}]})
    for item in result.get("output", []):
        for part in item.get("content", []):
            if part.get("type") == "output_text":
                print(part["text"])

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/responses", {"model": MODEL_ID,
        "input": "提取旅行信息：杭州，3 天。",
        "text": {"format": {"type": "json_schema", "name": "trip", "strict": True, "schema": SCHEMA}}})
    if result.get("status") != "completed":
        raise RuntimeError("响应尚未完成或已截断")
    texts = []
    for item in result.get("output", []):
        for part in item.get("content", []):
            if part.get("type") == "refusal":
                raise RuntimeError(part.get("refusal", "请求未完成"))
            if part.get("type") == "output_text":
                texts.append(part["text"])
    print(check_trip(json.loads("".join(texts))))

if __name__ == "__main__":
    main()