| 1234567891011121314151617181920212223242526272829303132 |
- import argparse
- import json
- from pathlib import Path
- def verify(path):
- # 只读取已保存的真实供应商响应与工具记录，不构造响应或再次调用模型。
- data = json.loads(path.read_text(encoding="utf-8"))
- assert data["status"] == "COMPLETED"
- assert 0 < data["actual_calls"] == len(data["attempts"]) <= data["request_limit"] == 3
- assert all(item["status"] == "RECEIVED" and item["usage"] for item in data["attempts"])
- calls = [call for message in data["messages"]
- for call in message["data"].get("tool_calls", [])]
- assert sum(call["name"] == "create_exchange_proposal" for call in calls) == 1
- assert any(call["name"] == "get_ticket" for call in calls)
- assert not any(call["name"] in {"confirm", "decide", "submit_exchange"} for call in calls)
- tickets = [json.loads(message["data"]["content"]) for message in data["messages"]
- if message["type"] == "tool"]
- assert tickets[-1]["status"] == "REJECTED" and tickets[-1]["operation_id"] is None
- assert tickets[-1]["ticket_id"] == tickets[0]["ticket_id"]
- assert data["messages"][-1]["type"] == "ai" and data["messages"][-1]["data"]["content"]
- print(json.dumps({
- "status": data["status"], "actual_calls": data["actual_calls"],
- "total_tokens": sum(item["usage"]["total_tokens"] for item in data["attempts"]),
- "ticket_status": tickets[-1]["status"], "additional_model_calls": 0,
- }))
- if __name__ == "__main__":
- parser = argparse.ArgumentParser(description="复核真实模型恢复运行的持久样本")
- parser.add_argument("path", type=Path)
- verify(parser.parse_args().path)
|
