L1 总览：把模型回答变成可检查的接口

"""读取客户求助，检查模型推荐的处理部门与理由。"""

import json
from dataclasses import dataclass
from typing import Callable, Dict, List, Optional

# 部门标识：Billing 为账单组，Technical 为技术支持组，General 为综合客服组。
TEAMS = {"Billing", "Technical", "General"}
CONTRACT = (
    "阅读客户求助记录（工单），推荐处理部门并说明理由，把建议交给客服人员确认。"
    "根据客户提供的内容判断。Billing 为账单组，Technical 为技术支持组，General 为综合客服组。"
    "输出 JSON 对象，字段恰好是 team、reason、missing_information。"
    "team 只能为 Billing、Technical、General 或 null；证据不足时为 null，并列出缺少的信息。"
)


@dataclass
class Result:
    status: str
    suggestion: Optional[Dict[str, object]]
    attempts: int
    trace: List[str]


def _unique_pairs(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate key: " + key)
        result[key] = value
    return result


def validate_response(raw: str) -> Dict[str, object]:
    """解析、检查字段和枚举；不声称判断业务事实是否正确。"""
    if not isinstance(raw, str):
        raise ValueError("response must be text")
    try:
        value = json.loads(raw, object_pairs_hook=_unique_pairs)
    except json.JSONDecodeError as error:
        raise ValueError("invalid_json") from error
    if not isinstance(value, dict) or set(value) != {"team", "reason", "missing_information"}:
        raise ValueError("fields must be exactly team, reason, missing_information")
    if value["team"] is not None and (not isinstance(value["team"], str) or value["team"] not in TEAMS):
        raise ValueError("team is not allowed")
    if not isinstance(value["reason"], str) or not value["reason"].strip():
        raise ValueError("reason must be non-empty text")
    missing = value["missing_information"]
    if not isinstance(missing, list) or any(not isinstance(item, str) or not item.strip() for item in missing):
        raise ValueError("missing_information must be a list of non-empty strings")
    if value["team"] is None and not missing:
        raise ValueError("missing_information is required when team is null")
    return value


def classify_ticket(
    ticket: Dict[str, str],
    transport: Callable[[List[Dict[str, str]]], str],
    max_attempts: int = 2,
) -> Result:
    """transport 是注入的只读响应函数；示例不会写入外部系统。"""
    title = ticket.get("title", "").strip()
    body = ticket.get("body", "").strip()
    if not title or not body:
        raise ValueError("title and body are required")
    if type(max_attempts) is not int or not 1 <= max_attempts <= 3:
        raise ValueError("max_attempts must be between 1 and 3")
    messages = [
        {"role": "developer", "content": CONTRACT},
        {"role": "user", "content": json.dumps({"title": title, "body": body}, ensure_ascii=False)},
    ]
    trace = []
    for attempt in range(1, max_attempts + 1):
        try:
            raw = transport(list(messages))
        except TimeoutError:
            trace.append("timeout")
            continue
        try:
            suggestion = validate_response(raw)
        except ValueError as error:
            category = "invalid_json" if str(error) == "invalid_json" else "invalid_schema"
            trace.append(category)
            messages.append({"role": "developer", "content": "上次输出未通过本地校验；仅重新给出符合契约的 JSON。"})
            continue
        return Result("needs_review" if suggestion["team"] is None else "suggested", suggestion, attempt, trace)
    status = "unknown" if trace and all(item == "timeout" for item in trace) else "needs_review"
    return Result(status, None, max_attempts, trace)


if __name__ == "__main__":
    class DemoTransport:
        def __call__(self, messages):
            return '{"team":null,"reason":"演示输入没有充分证据","missing_information":["分流规则"]}'

    example = classify_ticket({"title": "求助", "body": "帮我看看订单"}, DemoTransport())
    print(json.dumps(example.__dict__, ensure_ascii=False))

"""L1 综合实践：先规定可观察行为，再实现结构化调用器。"""

import json
import unittest

from pipeline import classify_ticket, validate_response


class FakeTransport:
    def __init__(self, answers):
        self.answers = list(answers)
        self.calls = []

    def __call__(self, messages):
        self.calls.append(messages)
        answer = self.answers.pop(0)
        if isinstance(answer, Exception):
            raise answer
        return answer


class PipelineTests(unittest.TestCase):
    def test_valid_answer_returns_suggestion_without_write(self):
        transport = FakeTransport(['{"team":"Billing","reason":"订单扣款后无记录","missing_information":[]}'])
        result = classify_ticket({"title": "账单", "body": "订单扣款后无记录"}, transport)
        self.assertEqual(result.status, "suggested")
        self.assertEqual(result.suggestion["team"], "Billing")
        self.assertEqual(result.attempts, 1)
        self.assertEqual(len(transport.calls), 1)

    def test_bad_json_can_be_retried_once_with_feedback(self):
        transport = FakeTransport(['不是 JSON', '{"team":null,"reason":"信息不足","missing_information":["订单号"]}'])
        result = classify_ticket({"title": "求助", "body": "请帮我看看"}, transport)
        self.assertEqual(result.status, "needs_review")
        self.assertEqual(result.attempts, 2)
        self.assertIn("invalid_json", result.trace)
        self.assertEqual(transport.calls[1][-1]["role"], "developer")

    def test_invalid_team_is_not_accepted(self):
        raw = '{"team":"Admin","reason":"我觉得应该处理","missing_information":[]}'
        with self.assertRaisesRegex(ValueError, "team"):
            validate_response(raw)

    def test_extra_fields_are_rejected(self):
        raw = '{"team":"Billing","reason":"扣款","missing_information":[],"write":true}'
        with self.assertRaisesRegex(ValueError, "fields"):
            validate_response(raw)

    def test_timeout_exhaustion_returns_unknown(self):
        transport = FakeTransport([TimeoutError("first"), TimeoutError("second")])
        result = classify_ticket({"title": "登录", "body": "无法登录"}, transport)
        self.assertEqual(result.status, "unknown")
        self.assertEqual(result.attempts, 2)
        self.assertEqual(result.trace, ["timeout", "timeout"])

    def test_blank_ticket_rejected_before_transport(self):
        transport = FakeTransport([])
        with self.assertRaisesRegex(ValueError, "title"):
            classify_ticket({"title": "", "body": "无法登录"}, transport)
        self.assertEqual(transport.calls, [])

    def test_duplicate_keys_are_rejected(self):
        raw = '{"team":"Billing","team":"Technical","reason":"冲突","missing_information":[]}'
        with self.assertRaisesRegex(ValueError, "duplicate"):
            validate_response(raw)

    def test_team_array_is_a_validation_error_not_a_crash(self):
        raw = '{"team":[],"reason":"没有依据","missing_information":[]}'
        with self.assertRaisesRegex(ValueError, "team"):
            validate_response(raw)

    def test_transport_wrong_type_is_recorded_as_invalid_response(self):
        transport = FakeTransport([None])
        result = classify_ticket({"title": "求助", "body": "无法登录"}, transport, max_attempts=1)
        self.assertEqual(result.status, "needs_review")
        self.assertEqual(result.trace, ["invalid_schema"])

    def test_boolean_attempt_limit_is_rejected_before_transport(self):
        transport = FakeTransport([])
        with self.assertRaisesRegex(ValueError, "max_attempts"):
            classify_ticket({"title": "求助", "body": "无法登录"}, transport, max_attempts=True)
        self.assertEqual(transport.calls, [])


if __name__ == "__main__":
    unittest.main()

dir
python -X utf8 -m unittest -v test_live_model.py

# 将占位文字替换为自己的教学账号密钥，仅在当前窗口设置。
$env:AGENTDEPTH_API_KEY = '你的教学账号密钥'
$env:AGENTDEPTH_PAID_CALL = 'YES'
try {
    python -X utf8 .\live_model.py
} finally {
    # 运行结束后清除当前窗口的密钥和调用开关。
    Remove-Item Env:AGENTDEPTH_API_KEY -ErrorAction SilentlyContinue
    Remove-Item Env:AGENTDEPTH_PAID_CALL -ErrorAction SilentlyContinue
}