baozaotumao2025
286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""Inspect a stuck execution and optionally probe target/judge calls without DB writes."""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core.config import get_settings
|
|
from app.db.models import TestResult, TestRun
|
|
from app.db.session import SessionLocal
|
|
from app.providers.model_provider import ProviderFactory
|
|
from app.services.dataset_service import DatasetGateway
|
|
from app.services.test_execution_service import TestExecutionService
|
|
|
|
|
|
def arguments():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--execution-id", default="R0169")
|
|
parser.add_argument("--run-id", type=int)
|
|
parser.add_argument(
|
|
"--probe",
|
|
action="store_true",
|
|
help="Actually call target and judge providers; never writes results",
|
|
)
|
|
parser.add_argument(
|
|
"--compare",
|
|
action="store_true",
|
|
help="Probe a cumulative schema ladder and stop at the first failure",
|
|
)
|
|
parser.add_argument("--timeout", type=float, default=60, help="Outer timeout per probe")
|
|
return parser.parse_args()
|
|
|
|
|
|
def retry_budget(timeout: float, retries: int, delay: float, max_delay: float) -> float:
|
|
return timeout * (retries + 1) + sum(
|
|
min(delay * 2**attempt, max_delay) for attempt in range(retries)
|
|
)
|
|
|
|
|
|
def elapsed_seconds(value: datetime) -> int:
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
return max(0, int((datetime.now(timezone.utc) - value).total_seconds()))
|
|
|
|
|
|
def find_run(db, run_id: int | None, execution_id: str):
|
|
if run_id:
|
|
return db.get(TestRun, run_id)
|
|
runs = db.scalars(select(TestRun).order_by(TestRun.id.desc())).all()
|
|
for run in runs:
|
|
summary = json.loads(run.summary_json or "{}")
|
|
if execution_id in {
|
|
summary.get("current_execution_id"),
|
|
summary.get("retry_execution_id"),
|
|
}:
|
|
return run
|
|
result = db.scalar(
|
|
select(TestResult)
|
|
.where(TestResult.execution_id == execution_id)
|
|
.order_by(TestResult.id.desc())
|
|
)
|
|
return db.get(TestRun, result.run_id) if result else None
|
|
|
|
|
|
def build_request(case):
|
|
model_input = case["model_input"]
|
|
messages = []
|
|
system_parts = [model_input["system"]] if model_input.get("system") else []
|
|
for attachment in model_input.get("attachments", []):
|
|
if attachment["type"] == "image":
|
|
messages.append(
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": attachment["file_content"]},
|
|
}
|
|
],
|
|
}
|
|
)
|
|
else:
|
|
system_parts.append("测试资料：" + json.dumps(attachment, ensure_ascii=False))
|
|
if system_parts:
|
|
messages.insert(0, {"role": "system", "content": "\n\n".join(system_parts)})
|
|
messages.extend(model_input.get("messages", []))
|
|
tools = [model_input["tool_schema"]] if model_input.get("tool_schema") else None
|
|
return messages, tools
|
|
|
|
|
|
def simple_tools(tools):
|
|
if not tools:
|
|
return None
|
|
function = tools[0]["function"]
|
|
return [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": function["name"],
|
|
"description": function.get("description", ""),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
def schema_ladder(tools):
|
|
simple = simple_tools(tools)
|
|
if not simple:
|
|
return []
|
|
original = tools[0]["function"]["parameters"]
|
|
variants = [("SIMPLE TOOL", simple)]
|
|
|
|
query_bounds = copy.deepcopy(simple)
|
|
query_bounds[0]["function"]["parameters"]["properties"]["query"] = copy.deepcopy(
|
|
original["properties"]["query"]
|
|
)
|
|
variants.append(("+ QUERY BOUNDS", query_bounds))
|
|
|
|
closed = copy.deepcopy(query_bounds)
|
|
closed[0]["function"]["parameters"]["additionalProperties"] = False
|
|
variants.append(("+ ADDITIONAL PROPERTIES FALSE", closed))
|
|
|
|
time_range = copy.deepcopy(closed)
|
|
parameters = time_range[0]["function"]["parameters"]
|
|
parameters["properties"]["time_range"] = copy.deepcopy(
|
|
original["properties"]["time_range"]
|
|
)
|
|
parameters["required"].append("time_range")
|
|
variants.append(("+ TIME RANGE ENUM", time_range))
|
|
|
|
source_enum = copy.deepcopy(time_range)
|
|
source_schema = copy.deepcopy(original["properties"]["source_types"])
|
|
source_schema.pop("uniqueItems", None)
|
|
source_schema.pop("maxItems", None)
|
|
source_enum[0]["function"]["parameters"]["properties"]["source_types"] = source_schema
|
|
variants.append(("+ SOURCE TYPES ENUM", source_enum))
|
|
|
|
variants.append(("+ ARRAY LIMITS (ORIGINAL)", tools))
|
|
return variants
|
|
|
|
|
|
async def timed(label, awaitable, timeout):
|
|
started = time.monotonic()
|
|
try:
|
|
value = await asyncio.wait_for(awaitable, timeout)
|
|
except Exception as exc:
|
|
print(f"{label}: FAIL {time.monotonic() - started:.2f}s {type(exc).__name__}: {exc}")
|
|
return None
|
|
print(f"{label}: OK {time.monotonic() - started:.2f}s")
|
|
return value
|
|
|
|
|
|
async def main():
|
|
args = arguments()
|
|
settings = get_settings()
|
|
case = next(
|
|
(
|
|
item
|
|
for item in DatasetGateway(settings).load()
|
|
if item["execution_id"] == args.execution_id
|
|
),
|
|
None,
|
|
)
|
|
if not case:
|
|
raise SystemExit(f"Unknown execution_id: {args.execution_id}")
|
|
|
|
with SessionLocal() as db:
|
|
run = find_run(db, args.run_id, args.execution_id)
|
|
if not run:
|
|
raise SystemExit("Run not found; pass --run-id explicitly")
|
|
summary = json.loads(run.summary_json or "{}")
|
|
result = db.scalar(
|
|
select(TestResult).where(
|
|
TestResult.run_id == run.id,
|
|
TestResult.execution_id == args.execution_id,
|
|
)
|
|
)
|
|
budget = retry_budget(
|
|
settings.request_timeout_seconds,
|
|
settings.request_retry_count,
|
|
settings.request_retry_delay_seconds,
|
|
settings.request_retry_max_delay_seconds,
|
|
)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"run_id": run.id,
|
|
"run_status": run.status,
|
|
"run_updated_seconds_ago": elapsed_seconds(run.updated_at),
|
|
"phase": summary.get("phase"),
|
|
"current_execution_id": summary.get("current_execution_id"),
|
|
"retry_execution_id": summary.get("retry_execution_id"),
|
|
"auto_judge": summary.get("auto_judge"),
|
|
"result": (
|
|
{
|
|
"execution_status": result.execution_status,
|
|
"verdict": result.verdict,
|
|
"error": result.error_message,
|
|
}
|
|
if result
|
|
else None
|
|
),
|
|
"case_mode": case.get("interaction_mode"),
|
|
"has_tool_schema": bool(case["model_input"].get("tool_schema")),
|
|
"provider_timeout_seconds": settings.request_timeout_seconds,
|
|
"provider_retry_count": settings.request_retry_count,
|
|
"worst_case_seconds_per_provider_call": budget,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
if result:
|
|
print("DIAGNOSIS: execution already has a stored result; refresh the run detail/UI.")
|
|
elif summary.get("current_execution_id") == args.execution_id:
|
|
print("DIAGNOSIS: backend is waiting in target or judge call, or its worker was interrupted.")
|
|
else:
|
|
print("DIAGNOSIS: execution is queued or the in-process background task was lost.")
|
|
if not args.probe and not args.compare:
|
|
print("NEXT: add --compare for target A/B tests or --probe for target and judge.")
|
|
return
|
|
|
|
factory = ProviderFactory(settings, db)
|
|
target = factory.create("target")
|
|
messages, tools = build_request(case)
|
|
print(f"TARGET: endpoint={target.endpoint} model={target.model}")
|
|
if args.compare:
|
|
outcomes = {}
|
|
for label, variant_tools in [("NO TOOLS", None), *schema_ladder(tools)]:
|
|
reply = await timed(
|
|
label, target.chat(messages, tools=variant_tools), args.timeout
|
|
)
|
|
outcomes[label] = "ok" if reply is not None else "timeout_or_error"
|
|
if reply is None:
|
|
break
|
|
print("A/B RESULT:", json.dumps(outcomes, ensure_ascii=False))
|
|
failed = next((label for label, result in outcomes.items() if result != "ok"), None)
|
|
if failed == "NO TOOLS":
|
|
print("DIAGNOSIS: prompt/model/server path is slow; tool schema is not the cause.")
|
|
elif failed:
|
|
print(f"DIAGNOSIS: first failing schema stage is {failed}.")
|
|
else:
|
|
print("DIAGNOSIS: the full original schema works now; suspect intermittent load.")
|
|
return
|
|
reply = await timed("TARGET", target.chat(messages, tools=tools), args.timeout)
|
|
if reply is None:
|
|
return
|
|
response = reply.content or json.dumps(
|
|
{"tool_calls": reply.tool_calls}, ensure_ascii=False
|
|
)
|
|
print(
|
|
"TARGET RESULT:",
|
|
json.dumps(
|
|
{
|
|
"content_length": len(reply.content),
|
|
"tool_call_count": len(reply.tool_calls),
|
|
"preview": response[:500],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
if summary.get("auto_judge"):
|
|
judge = factory.create("judge")
|
|
print(f"JUDGE: endpoint={judge.endpoint} model={judge.model}")
|
|
service = TestExecutionService(db=db, target_provider=target, judge_provider=judge)
|
|
judged = await timed("JUDGE", service._judge(case, response), args.timeout)
|
|
if judged is not None:
|
|
print("JUDGE RESULT:", json.dumps(judged, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|