88 lines
5.5 KiB
Python
88 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
||
"""Cost-first OpenAI-compatible model router with a macOS Keychain secret reference."""
|
||
from __future__ import annotations
|
||
import argparse, hashlib, json, subprocess
|
||
from pathlib import Path
|
||
import urllib.error, urllib.request
|
||
from typing import Any
|
||
|
||
ROOT=Path(__file__).resolve().parents[2]
|
||
CONFIG=ROOT/"routing/persona-model-api-smart-router-map.json"
|
||
|
||
class RouterError(RuntimeError): pass
|
||
|
||
def load(path: Path) -> dict[str,Any]:
|
||
value=json.loads(path.read_text())
|
||
if not isinstance(value,dict): raise RouterError("JSON_OBJECT_REQUIRED")
|
||
return value
|
||
|
||
def keychain_secret(config: dict[str,Any]) -> str:
|
||
p=config["provider"]
|
||
result=subprocess.run(["security","find-generic-password","-a",p["keychain_account"],"-s",p["keychain_service"],"-w"],text=True,capture_output=True,timeout=10)
|
||
if result.returncode or not result.stdout.strip(): raise RouterError("OPENLUX_KEYCHAIN_SECRET_UNAVAILABLE")
|
||
return result.stdout.strip()
|
||
|
||
def request_json(url: str, key: str, payload: dict[str,Any]|None=None, timeout: int=45) -> dict[str,Any]:
|
||
if not url.startswith("https://api.openlux.ai/v1/"): raise RouterError("PROVIDER_URL_NOT_ALLOWLISTED")
|
||
body=None if payload is None else json.dumps(payload,ensure_ascii=False).encode()
|
||
request=urllib.request.Request(url,data=body,method="GET" if body is None else "POST",headers={"Authorization":f"Bearer {key}","Content-Type":"application/json"})
|
||
try:
|
||
with urllib.request.urlopen(request,timeout=timeout) as response:
|
||
value=json.loads(response.read())
|
||
except urllib.error.HTTPError as error:
|
||
raise RouterError(f"MODEL_HTTP_{error.code}") from error
|
||
except Exception as error:
|
||
raise RouterError(f"MODEL_TRANSPORT_{type(error).__name__}") from error
|
||
if not isinstance(value,dict): raise RouterError("MODEL_RESPONSE_OBJECT_REQUIRED")
|
||
return value
|
||
|
||
def available_models(config: dict[str,Any], key: str) -> set[str]:
|
||
value=request_json(config["provider"]["base_url"]+"/models",key,timeout=20)
|
||
return {x.get("id") for x in value.get("data",[]) if isinstance(x,dict) and isinstance(x.get("id"),str)}
|
||
|
||
def candidates(config: dict[str,Any], complexity: str, available: set[str]) -> list[str]:
|
||
order=["LOW","MEDIUM","HIGH"]
|
||
level=complexity if complexity in order else config["policy"]["default_complexity"]
|
||
result=[]
|
||
for tier in order[:order.index(level)+1]:
|
||
result.extend(x for x in config["tiers"][tier] if x in available and x not in result)
|
||
return result[:int(config["policy"]["max_attempts"])]
|
||
|
||
def route(event: dict[str,Any], config: dict[str,Any]|None=None, secret_reader=keychain_secret) -> dict[str,Any]:
|
||
config=config or load(CONFIG)
|
||
for field in config["identity_envelope_required"]:
|
||
if not isinstance(event.get(field),str) or not event[field].strip(): raise RouterError(f"IDENTITY_ENVELOPE_REQUIRED:{field}")
|
||
key=secret_reader(config)
|
||
available=available_models(config,key)
|
||
selected=candidates(config,event.get("complexity","LOW"),available)
|
||
if not selected: raise RouterError("NO_CONFIGURED_MODEL_AVAILABLE")
|
||
system=(f"你是{event['parent_persona_id']}人格系统内的子Agent {event['child_agent_id']}，岗位是{event['agent_role']}。"
|
||
f"你的唯一父人格主控是{event['controller']}，当前频道{event['channel_id']}。权限仅为{event['authority']}。"
|
||
"你必须理解后再协作，不能把宿主规则程序冒充人格，不能扩大权限，不能泄露私人数据。"
|
||
f"你已领取的TCS共享认知脑是：{json.dumps(event.get('shared_cognition',{}),ensure_ascii=False)}")
|
||
attempts=[]
|
||
for model in selected:
|
||
try:
|
||
response=request_json(config["provider"]["base_url"]+"/chat/completions",key,{"model":model,"stream":False,"temperature":0.2,"messages":[{"role":"system","content":system},{"role":"user","content":event["task"]}]})
|
||
content=response.get("choices",[{}])[0].get("message",{}).get("content")
|
||
if not isinstance(content,str) or not content.strip(): raise RouterError("MODEL_RESPONSE_EMPTY")
|
||
return {"outcome":"PASS","router_id":config["router_id"],"selected_model":model,"attempts":[*attempts,{"model":model,"outcome":"PASS"}],"response":content,"response_sha256":hashlib.sha256(content.encode()).hexdigest(),"identity_bound":True,"credentials_stored":False,"authority_granted":False}
|
||
except RouterError as error:
|
||
attempts.append({"model":model,"outcome":"FAIL","error_code":str(error)})
|
||
raise RouterError("ALL_COST_ORDERED_MODELS_FAILED:"+",".join(x["model"] for x in attempts))
|
||
|
||
def main() -> int:
|
||
p=argparse.ArgumentParser(); p.add_argument("command",choices=("probe","route")); p.add_argument("--input"); a=p.parse_args()
|
||
try:
|
||
config=load(CONFIG)
|
||
if a.command=="probe":
|
||
key=keychain_secret(config); models=available_models(config,key); chosen=candidates(config,"LOW",models)
|
||
result={"outcome":"PASS","router_id":config["router_id"],"provider":config["provider"]["id"],"available_model_count":len(models),"low_cost_candidates":chosen,"keychain_reference":True,"secret_exposed":False}
|
||
else:
|
||
if not a.input: raise RouterError("INPUT_REQUIRED")
|
||
result=route(load(Path(a.input)),config)
|
||
print(json.dumps(result,ensure_ascii=False,indent=2,sort_keys=True)); return 0
|
||
except Exception as error:
|
||
print(json.dumps({"outcome":"FAIL","error":str(error),"secret_exposed":False},ensure_ascii=False)); return 2
|
||
|
||
if __name__=="__main__": raise SystemExit(main())
|