Google · Gemini 原生聊天

# 安装依赖：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)
        if event == "error" or item.get("error"):
            raise RuntimeError(item)
        if item.get("promptFeedback", {}).get("blockReason"):
            raise RuntimeError(item["promptFeedback"])
        # 本示例读取第一个候选；多候选请求需按 index 分别累积。
        candidates = item.get("candidates", [])
        for candidate in candidates:
            if candidate.get("index", 0) != 0:
                continue
            for part in candidate.get("content", {}).get("parts", []):
                if "text" in part and not part.get("thought", False):
                    yield part["text"]
            reason = candidate.get("finishReason")
            if reason:
                if reason != "STOP":
                    raise RuntimeError("输出未完整完成：" + reason)
                completed = True
        # 原生 SSE 不要求 [DONE]；STOP 后读到正常 EOF 即可结束。
    if not completed:
        raise RuntimeError("流结束但没有收到候选的 STOP 终止原因。")

def main():
    path = "/v1beta/models/" + quote(MODEL_ID, safe="") + ":streamGenerateContent?alt=sse"
    body = {"contents": [{"role": "user", "parts": [{"text": "用三句话介绍杭州。"}]}]}
    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():
    path = "/v1beta/models/" + quote(MODEL_ID, safe="") + ":generateContent"
    # Gemini 的 Schema 子集按所选模型填写。
    parameters = {"type": "OBJECT", "properties": {
        "a": {"type": "NUMBER"}, "b": {"type": "NUMBER"}}, "required": ["a", "b"]}
    tools = [{"functionDeclarations": [{"name": "add_numbers",
              "description": "计算两个数字之和", "parameters": parameters}]}]
    contents = [{"role": "user", "parts": [{"text": "请用工具计算 2 加 3。"}]}]
    first = request_json(path, {"contents": contents, "tools": tools,
        "toolConfig": {"functionCallingConfig": {"mode": "AUTO"}}})
    model_content = first["candidates"][0]["content"]
    calls = [part["functionCall"] for part in model_content["parts"] if "functionCall" in part]
    if not calls:
        print(first)
        return
    contents.append(model_content)  # 原样保留 parts，包括 thoughtSignature。
    parts = []
    for call in calls:
        result = run_tool(call["name"], call["args"])
        response = {"name": call["name"], "response": result}
        if "id" in call:
            response["id"] = call["id"]
        parts.append({"functionResponse": response})
    contents.append({"role": "user", "parts": parts})
    second = request_json(path, {"contents": contents, "tools": tools,
        "toolConfig": {"functionCallingConfig": {"mode": "NONE"}}})
    for part in second["candidates"][0]["content"]["parts"]:
        if "text" in part and not part.get("thought", False):
            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")
    path = "/v1beta/models/" + quote(MODEL_ID, safe="") + ":generateContent"
    result = request_json(path, {"contents": [{"role": "user", "parts": [
        {"text": "请描述这张图片。"},
        {"inlineData": {"mimeType": "image/png", "data": image}}
    ]}]})
    for part in result["candidates"][0]["content"]["parts"]:
        if "text" in part and not part.get("thought", False):
            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():
    path = "/v1beta/models/" + quote(MODEL_ID, safe="") + ":generateContent"
    result = request_json(path, {
        "contents": [{"role": "user", "parts": [{"text": "提取旅行信息：杭州，3 天。"}]}],
        "generationConfig": {"responseMimeType": "application/json", "responseSchema": {
            "type": "OBJECT", "properties": {"city": {"type": "STRING"}, "days": {"type": "INTEGER"}},
            "required": ["city", "days"]}}})
    candidate = result["candidates"][0]
    if candidate.get("finishReason") != "STOP":
        raise RuntimeError("未得到完整结构化结果")
    text = "".join(part.get("text", "") for part in candidate["content"]["parts"]
                   if not part.get("thought", False))
    print(check_trip(json.loads(text)))

if __name__ == "__main__":
    main()