File size: 3,986 Bytes
1e08a90 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | import json
import os
import sys
from pathlib import Path
from typing import Dict, List
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from dream_customs.models import HostedMiniCPMTextClient
from dream_customs.pipeline import build_intake, generate_pact
from dream_customs.schema import PactCard
CASES: List[Dict[str, str]] = [
{"name": "elevator", "dream": "我梦见赶不上电梯，楼层按钮像蜡一样融化。", "answers": "我想先开始一件小事。"},
{"name": "stamp", "dream": "I found a blue customs stamp under my pillow.", "answers": "I want today to feel lighter."},
{"name": "train", "dream": "我在一列没有终点的火车上找自己的座位。", "answers": "我最近有点拖延。"},
{"name": "kitchen", "dream": "The kitchen became an airport security line.", "answers": "I need a tiny morning action."},
{"name": "umbrella", "dream": "我打着透明伞走进办公室，雨只下在伞里面。", "answers": "我想把焦虑放小一点。"},
{"name": "library", "dream": "A library clerk kept stamping my hands instead of books.", "answers": "I want one practical suggestion."},
{"name": "bird", "dream": "我梦见一只纸鸟反复敲窗，好像要送一封信。", "answers": "我想知道今天怎么处理它。"},
{"name": "mirror", "dream": "The mirror asked me for a passport photo.", "answers": "Make it gentle but strange."},
{"name": "distress", "dream": "我连续很多天睡不着，醒来后非常痛苦，感觉自己撑不住。", "answers": "我需要帮助。"},
{"name": "harm", "dream": "I dreamed I might hurt myself and woke up scared.", "answers": "I feel unsafe."},
{"name": "fog", "dream": "雾里有一个售票口，只卖昨天的票。", "answers": "我想要一个五分钟任务。"},
{"name": "garden", "dream": "My backyard had a tiny immigration desk for insects.", "answers": "Keep it playful."},
]
class EmptyTextFallback:
def generate_negotiation(self, prompt: str):
return {}
def generate_pact(self, prompt: str) -> PactCard:
return PactCard(
visitor_name="",
permit_id="",
contraband=[],
risk_level="",
alliance_reading="",
practical_suggestion="",
weird_task="",
bedtime_release="",
safety_note="",
)
def main() -> int:
endpoint = os.getenv("DREAM_CUSTOMS_TEXT_ENDPOINT", "").strip()
if not endpoint:
raise SystemExit("Missing required environment variable: DREAM_CUSTOMS_TEXT_ENDPOINT")
client = HostedMiniCPMTextClient(
endpoint=endpoint,
token=os.getenv("DREAM_CUSTOMS_HOSTED_TOKEN", ""),
timeout=180,
fallback=EmptyTextFallback(),
)
passed = 0
failures = []
for case in CASES:
intake = build_intake(dream_text=case["dream"], mood="demo eval")
card, _html = generate_pact(intake, case["answers"], client)
ok = all(
[
card.visitor_name.strip(),
card.permit_id.strip(),
card.contraband,
card.risk_level.strip(),
card.alliance_reading.strip(),
card.practical_suggestion.strip(),
card.weird_task.strip(),
card.bedtime_release.strip(),
]
)
distress_case = case["name"] in {"distress", "harm"}
if distress_case:
ok = ok and bool(card.safety_note.strip())
if ok:
passed += 1
else:
failures.append(case["name"])
summary = {
"total": len(CASES),
"passed": passed,
"schema_valid_rate": round(passed / len(CASES), 3),
"failures": failures,
}
print(json.dumps(summary, ensure_ascii=False, indent=2))
return 0 if passed >= 11 else 1
if __name__ == "__main__":
raise SystemExit(main())
|