bingshuo/guanghu-ice-heart
Watch
1
0
Fork
You've already forked guanghu-ice-heart
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

feat(heartbeat): add four-host HoloLake control console

Browse source
This commit is contained in:
冰朔 2026-09-09 19:45:20 +08:00
commit 5ca2c9d66f
39 changed files with 1784 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

3
server-tools/persona-model-smart-router/README.md
View file

@ -3,3 +3,6 @@
The router uses the OpenAI-compatible endpoint at `https://api.openlux.ai/v1`. Its API key is read at call time from macOS Keychain service `ai.openlux.api`, account `guanghu-persona-agent-router`; no secret is stored in this repository.
Each call must carry the active parent persona, derived child Agent ID, controller, channel, job role, bounded authority and task. The router tries the configured low-cost healthy models first and escalates only after a transport, empty-response or downstream validation failure. Provider pricing is not exposed by the current `/models` response, so the configured order is explicitly a heuristic rather than a verified price table.
The default remains `COST_FIRST`. An explicitly registered reasoning advisor may send
`selection_policy=QUALITY_FIRST` with `complexity=HIGH`; this reverses the eligible tier order so a high-tier model is tried first. It does not grant execution, write, approval, or persona-cognition authority.

14
server-tools/persona-model-smart-router/openlux_router.py
View file

@ -40,11 +40,14 @@ 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]:
def candidates(config: dict[str,Any], complexity: str, available: set[str], selection_policy: str="COST_FIRST") -> list[str]:
order=["LOW","MEDIUM","HIGH"]
level=complexity if complexity in order else config["policy"]["default_complexity"]
tiers=order[:order.index(level)+1]
if selection_policy == "QUALITY_FIRST": tiers=list(reversed(tiers))
elif selection_policy != "COST_FIRST": raise RouterError("SELECTION_POLICY_UNSUPPORTED")
result=[]
for tier in order[:order.index(level)+1]:
for tier in tiers:
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"])]
@ -54,11 +57,14 @@ def route(event: dict[str,Any], config: dict[str,Any]|None=None, secret_reader=k
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)
selection_policy=event.get("selection_policy","COST_FIRST")
selected=candidates(config,event.get("complexity","LOW"),available,selection_policy)
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']}。"
"你必须理解后再协作，不能把宿主规则程序冒充人格，不能扩大权限，不能泄露私人数据。"
"你只能把输入中明确提供的内容称为事实且必须标为FACT；推断必须标为INFERENCE，缺证据必须标为UNKNOWN。"
"禁止虚构日志、错误码、编号、回执、人物回应或系统状态；模型返回不等于已校验、已执行或已完成。"
f"你已领取的TCS共享认知脑是：{json.dumps(event.get('shared_cognition',{}),ensure_ascii=False)}")
attempts=[]
for model in selected:
@ -66,7 +72,7 @@ def route(event: dict[str,Any], config: dict[str,Any]|None=None, secret_reader=k
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}
return {"outcome":"PASS","router_id":config["router_id"],"selected_model":model,"selection_policy":selection_policy,"attempts":[*attempts,{"model":model,"outcome":"PASS"}],"response":content,"response_sha256":hashlib.sha256(content.encode()).hexdigest(),"response_validation_state":"CALLER_REVIEW_REQUIRED","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))

9
server-tools/persona-model-smart-router/test_openlux_router.py
View file

@ -9,14 +9,21 @@ E={"parent_persona_id":"ICE-P-ZY001","child_agent_id":"SUBAGENT::ICE-P-ZY001::TE
class Tests(unittest.TestCase):
def test_cost_order_uses_only_available_models(self):
self.assertEqual(M.candidates(C,'LOW',{'glm-4-flash','gpt-5-nano'}),['glm-4-flash','gpt-5-nano'])
def test_quality_first_uses_high_tier_before_lower_tiers(self):
available={'qwen-flash','gpt-5.4-mini','gpt-6-astra-2026-09-03'}
self.assertEqual(M.candidates(C,'HIGH',available,'QUALITY_FIRST'),['gpt-6-astra-2026-09-03','gpt-5.4-mini','qwen-flash'])
def test_identity_envelope_and_key_never_returned(self):
def fake(url,key,payload=None,timeout=45):
if url.endswith('/models'): return {"data":[{"id":"qwen-flash"}]}
self.assertIn('ICE-P-ZY001',payload['messages'][0]['content']); self.assertEqual(key,'hidden')
self.assertIn('ICE-P-ZY001',payload['messages'][0]['content'])
self.assertIn('FACT',payload['messages'][0]['content'])
self.assertIn('禁止虚构日志',payload['messages'][0]['content'])
self.assertEqual(key,'hidden')
return {"choices":[{"message":{"content":"收到"}}]}
with patch.object(M,'request_json',side_effect=fake):
result=M.route(E,C,secret_reader=lambda _: 'hidden')
self.assertEqual(result['selected_model'],'qwen-flash'); self.assertNotIn('hidden',json.dumps(result))
self.assertEqual(result['response_validation_state'],'CALLER_REVIEW_REQUIRED')
def test_missing_identity_and_unallowlisted_url_fail(self):
with self.assertRaisesRegex(M.RouterError,'IDENTITY_ENVELOPE'): M.route({"task":"x"},C,secret_reader=lambda _:'hidden')
with self.assertRaisesRegex(M.RouterError,'NOT_ALLOWLISTED'): M.request_json('https://evil.example/v1/models','hidden')