Jev 爆红之后：判断模型真正的护城河不在架构，而在数据闭环

用户请求
  ↓
主模型生成多个候选结果
  ↓
判断模型评分或选出胜者
  ↓
返回结果、重试或进入人工审核

{
  "task": "比较两个候选回答",
  "input": "用户问题",
  "candidate_a": "候选回答 A",
  "candidate_b": "候选回答 B",
  "result": {
    "winner": "A",
    "reason": "A 给出了正确且可执行的步骤"
  }
}

{"question":"计算 17 × 6。","answer_a":"102","answer_b":"96","expected":"A"}
{"question":"Python 中如何安全读取可能不存在的环境变量？","answer_a":"使用 os.environ['KEY']，不存在时程序会抛出异常。","answer_b":"使用 os.getenv('KEY')，并为缺失值提供默认值或显式检查。","expected":"B"}

import json
import os
import urllib.request

ENDPOINT = os.environ.get(
    'JUDGE_ENDPOINT',
    'https://api.example.com/v1/chat/completions',
)
MODEL = os.environ.get('JUDGE_MODEL', 'replace-with-your-judge-model')
TOKEN = os.environ.get('JUDGE_TOKEN', '')


def call_judge(question, answer_a, answer_b):
    prompt = f'''You are an impartial answer judge.
Compare the two answers for correctness, relevance, and completeness.
Do not prefer an answer because of its position or writing style.
Return JSON only: {{"winner":"A|B|TIE","reason":"short reason"}}.

Question: {question}

Answer A: {answer_a}

Answer B: {answer_b}
'''

    payload = {
        'model': MODEL,
        'temperature': 0,
        'response_format': {'type': 'json_object'},
        'messages': [{'role': 'user', 'content': prompt}],
    }
    headers = {'Content-Type': 'application/json'}
    if TOKEN:
        headers['Authorization'] = f'Bearer {TOKEN}'

    request = urllib.request.Request(
        ENDPOINT,
        data=json.dumps(payload).encode('utf-8'),
        headers=headers,
        method='POST',
    )
    with urllib.request.urlopen(request, timeout=60) as response:
        body = json.load(response)

    content = body['choices'][0]['message']['content']
    return json.loads(content)


def reverse_winner(winner):
    return {'A': 'B', 'B': 'A', 'TIE': 'TIE'}[winner]


def main():
    total = 0
    correct = 0
    consistent = 0

    with open('eval.jsonl', encoding='utf-8') as source:
        for line in source:
            if not line.strip():
                continue
            item = json.loads(line)

            normal = call_judge(
                item['question'], item['answer_a'], item['answer_b']
            )
            swapped = call_judge(
                item['question'], item['answer_b'], item['answer_a']
            )
            swapped_in_original_order = reverse_winner(swapped['winner'])

            total += 1
            correct += normal['winner'] == item['expected']
            consistent += normal['winner'] == swapped_in_original_order

            print(json.dumps({
                'question': item['question'],
                'expected': item['expected'],
                'winner': normal['winner'],
                'reason': normal['reason'],
                'winner_after_swap': swapped_in_original_order,
            }, ensure_ascii=False))

    print(json.dumps({
        'samples': total,
        'accuracy': correct / total if total else 0,
        'swap_consistency': consistent / total if total else 0,
    }, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()

export JUDGE_ENDPOINT='https://your-provider.example/v1/chat/completions'
export JUDGE_MODEL='your-judge-model'
export JUDGE_TOKEN='your-api-token'
python judge_eval.py