107 lines
7.1 KiB
Python
107 lines
7.1 KiB
Python
"""Four isolated Gemini measurements; default prepares only, --run makes four calls."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from jsonschema import Draft202012Validator
|
|
from .llm import GatewayConfig, LLMError
|
|
from .llm.gateway import PRIVACY
|
|
from .llm.gemini import GeminiProvider
|
|
from .llm.diagnostics import count
|
|
from .telemetry.safety import safe_code
|
|
|
|
SCHEMA = {"type":"object","properties":{"step":{"type":"string"},"purpose":{"type":"string"}},
|
|
"required":["step","purpose"],"additionalProperties":False}
|
|
PROMPTS = {
|
|
"de": "Bug: Beim Braufest erscheinen vor Eisenschmiede keine Wolpertinger, obwohl der Spieler die erforderliche Brille trägt. In Orgrimmar funktioniert die entsprechende Quest. Hypothese: In der Datenbank fehlen Wolpertinger-Spawns für Dun Morogh. Welchen einzelnen Untersuchungsschritt würdest du als Nächstes durchführen? Antworte knapp auf Deutsch mit der benötigten Abfrage und ihrem Zweck. Du kannst hier keine Werkzeuge ausführen.",
|
|
"en": "Bug: During Brewfest, no Wolpertingers appear outside Ironforge although the player wears the required goggles. The corresponding quest works in Orgrimmar. Hypothesis: The database lacks Wolpertinger spawns for Dun Morogh. Which single investigation step would you perform next? Reply briefly in English with the required query and its purpose. You cannot execute tools here.",
|
|
}
|
|
|
|
|
|
def make_plan(provider):
|
|
cases=[]
|
|
for language,prompt in PROMPTS.items():
|
|
for structured in (False,True):
|
|
generation={"maxOutputTokens":provider.max_output_tokens}
|
|
if provider.reasoning_effort is not None:
|
|
generation["thinkingConfig"]={"thinkingLevel":provider.reasoning_effort}
|
|
if structured:
|
|
generation.update(responseMimeType="application/json",responseJsonSchema=SCHEMA)
|
|
cases.append({"id":language+("_json" if structured else "_plain"),
|
|
"prompt_characters":len(prompt),"payload":{"contents":[{"role":"user","parts":[{"text":prompt}]}],
|
|
"generationConfig":generation}})
|
|
return {"format":"viano-minimal-prompt-test-v1","provider":provider.id,"model":provider.model,
|
|
"explicit_cache":False,"automatic_cache":"provider controlled; usage reported",
|
|
"calls":cases,"notes":"No system prompt, history, case data, tools, retries, failover or investigation wrapper. JSON variants differ only in native generation format/schema."}
|
|
|
|
|
|
def measure(provider,case,limit,post=None):
|
|
secret=os.environ.get(provider.api_key_env) if provider.api_key_env else None
|
|
if provider.api_key_env and (not secret or any(ord(c)<32 or ord(c)==127 for c in secret)):
|
|
raise LLMError("LLM_CREDENTIALS")
|
|
headers={"Content-Type":"application/json","Accept":"application/json"}
|
|
if secret: headers["x-goog-api-key"]=secret
|
|
start=time.monotonic()
|
|
result={"id":case["id"],"prompt_characters":case["prompt_characters"]}
|
|
try:
|
|
envelope=(post or GeminiProvider._post)(provider.base_url.rstrip('/')+'/models/'+provider.model+':generateContent',
|
|
case['payload'],headers,limit,start+provider.timeout_seconds)
|
|
usage=envelope.get('usageMetadata') or {}
|
|
result['usage']={name:count(usage.get(key)) for name,key in {
|
|
'input_tokens':'promptTokenCount','candidate_output_tokens':'candidatesTokenCount',
|
|
'reasoning_tokens':'thoughtsTokenCount','cached_input_tokens':'cachedContentTokenCount',
|
|
'total_tokens':'totalTokenCount'}.items()}
|
|
candidate=envelope['candidates'][0]
|
|
text=''.join(part['text'] for part in candidate.get('content',{}).get('parts',[]) if 'text' in part and not part.get('thought'))
|
|
result.update(answer=text,finish_reason=candidate.get('finishReason'),model=envelope.get('modelVersion',provider.model))
|
|
result['status']='SUCCESS' if candidate.get('finishReason')=='STOP' and text else 'INCOMPLETE'
|
|
if ('validation_schema' in case or 'responseJsonSchema' in case['payload']['generationConfig']) and result['status']=='SUCCESS':
|
|
try: valid=Draft202012Validator(case['validation_schema'] if 'validation_schema' in case else case['payload']['generationConfig']['responseJsonSchema']).is_valid(json.loads(text))
|
|
except ValueError: valid=False
|
|
result['schema_valid']=valid
|
|
if not valid: result['status']='INVALID_JSON'
|
|
except LLMError as error:
|
|
result.update(status='ERROR',code=safe_code(error.code),http_status=error.http_status)
|
|
except (ValueError,TypeError,KeyError,IndexError,AttributeError):
|
|
result.update(status='ERROR',code='LLM_RESPONSE')
|
|
result['elapsed_seconds']=round(time.monotonic()-start,3)
|
|
return result
|
|
|
|
|
|
def main(argv=None):
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--llm-config',default='config/llm.working-brief.local.toml')
|
|
parser.add_argument('--provider',help='Optional exact Gemini provider ID; otherwise first eligible provider')
|
|
parser.add_argument('--output',type=Path,default=Path('service-data/benchmarks/minimal-prompts-20261017'))
|
|
parser.add_argument('--run',action='store_true',help='Make four potentially billable calls; default prepares only')
|
|
args=parser.parse_args(argv)
|
|
try:
|
|
config=GatewayConfig.load(args.llm_config)
|
|
eligible=[p for p in config.providers if p.enabled and p.privacy in PRIVACY[config.privacy_profile]]
|
|
provider=next((p for p in eligible if p.id==args.provider),None) if args.provider else next(iter(eligible),None)
|
|
if provider is None or provider.adapter!='gemini':raise LLMError('MINIMAL_TEST_PROVIDER')
|
|
plan=make_plan(provider);args.output.mkdir(parents=True,exist_ok=True)
|
|
plan_path=args.output/'plan.json';result_path=args.output/'results.json'
|
|
if plan_path.exists() and json.loads(plan_path.read_text(encoding='utf-8'))!=plan:
|
|
raise LLMError('MINIMAL_TEST_PLAN_CHANGED')
|
|
if not plan_path.exists():plan_path.write_text(json.dumps(plan,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
|
|
if not args.run:
|
|
print(json.dumps({'status':'PREPARED','plan':str(plan_path),'calls':4,'network_used':False}));return 0
|
|
# Reserve result file before the first call; never silently repeat a paid run.
|
|
with result_path.open('x',encoding='utf-8') as f:json.dump({'status':'RUNNING','results':[]},f)
|
|
results=[]
|
|
for case in plan['calls']:
|
|
result=measure(provider,case,config.max_response_bytes);results.append(result)
|
|
result_path.write_text(json.dumps({'status':'COMPLETE' if len(results)==4 else 'RUNNING','results':results},ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
|
|
print(json.dumps(result,ensure_ascii=True),flush=True)
|
|
return 0 if all(r['status']=='SUCCESS' for r in results) else 2
|
|
except KeyboardInterrupt:
|
|
print(json.dumps({'status':'INTERRUPTED'}));return 130
|
|
except (LLMError,OSError,ValueError) as error:
|
|
print(json.dumps({'status':'ERROR','code':safe_code(getattr(error,'code',None))}));return 2
|
|
|
|
|
|
if __name__=='__main__':raise SystemExit(main())
|