示例与测试代码

import json
import time
import requests

API_KEY = "kct_your_api_key"
BASE_URL = "https://keath.ai/api/keath/public/v1"


def get_credits():
    response = requests.get(
        f"{BASE_URL}/credits",
        headers={"X-API-Key": API_KEY},
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


def preview_questions_with_pdf(path: str):
    with open(path, "rb") as f:
        response = requests.post(
            f"{BASE_URL}/questions-ingest-preview",
            headers={"X-API-Key": API_KEY},
            files={"file": ("question-pack.pdf", f, "application/pdf")},
            data={"specification": "只解析 Question 2，忽略 sample answer 页面。"},
            timeout=300,
        )
    response.raise_for_status()
    return response.json()


def preview_questions_mixed(path: str):
    with open(path, "rb") as f:
        response = requests.post(
            f"{BASE_URL}/questions-ingest-preview",
            headers={"X-API-Key": API_KEY},
            files={"file": ("scan-1.jpg", f, "image/jpeg")},
            data={
                "assets": json.dumps(
                    [
                        {
                            "url": "https://cdn.example.com/supporting-page.pdf",
                            "mime_type": "application/pdf",
                            "name": "supporting-page.pdf",
                        }
                    ]
                ),
                "specification": "以上传图片为目标题目。",
            },
            timeout=300,
        )
    response.raise_for_status()
    return response.json()


def submit_evaluation(assignment_id: int, paper_content: str):
    response = requests.post(
        f"{BASE_URL}/evaluations",
        headers={
            "X-API-Key": API_KEY,
            "Content-Type": "application/json",
        },
        json={
            "assignment_id": assignment_id,
            "paper_content": paper_content,
            "style": "Bullet",
        },
        timeout=300,
    )
    response.raise_for_status()
    return response.json()


def submit_evaluation_with_file(assignment_id: int, answer_path: str):
    with open(answer_path, "rb") as f:
        response = requests.post(
            f"{BASE_URL}/evaluations",
            headers={"X-API-Key": API_KEY},
            data={
                "assignment_id": str(assignment_id),
                "student_id": "anon-student-001",
                "style": "Bullet",
            },
            files={"answer_file": ("student-answer.pdf", f, "application/pdf")},
            timeout=300,
        )
    response.raise_for_status()
    return response.json()


def submit_one_pass_evaluation(
    model_id: int,
    question_path: str,
    rubric_path: str,
    answer_path: str,
    request_id: str,
):
    with open(question_path, "rb") as question, open(rubric_path, "rb") as rubric, open(answer_path, "rb") as answer:
        response = requests.post(
            f"{BASE_URL}/evaluations/one-pass",
            headers={
                "X-API-Key": API_KEY,
                "Idempotency-Key": request_id,
            },
            data={
                "model_id": str(model_id),
                "student_id": "anon-student-001",
                "specification": "只使用 Question 2。",
                "style": "Bullet",
            },
            files={
                "question_file": ("question.pdf", question, "application/pdf"),
                "rubric_file": (
                    "rubric.docx",
                    rubric,
                    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
                ),
                "answer_file": ("student-answer.txt", answer, "text/plain"),
            },
            timeout=300,
        )
    response.raise_for_status()
    return response.json()


def wait_for_evaluation(task_id: str):
    max_attempts = 20
    initial_delay_seconds = 3
    max_delay_seconds = 15

    for attempt in range(1, max_attempts + 1):
        response = requests.get(
            f"{BASE_URL}/evaluations/{task_id}/status",
            headers={"X-API-Key": API_KEY},
            timeout=60,
        )
        response.raise_for_status()
        payload = response.json()
        data = payload.get("data", payload)
        status = data["status"]
        if status in {"SUCCESS", "FAILURE", "CANCEL"}:
            return payload
        delay_seconds = min(initial_delay_seconds * attempt, max_delay_seconds)
        time.sleep(delay_seconds)

    raise TimeoutError(
        f"评测任务 {task_id} 在 {max_attempts} 次轮询后仍未结束"
    )