203 lines
8.3 KiB
Python
203 lines
8.3 KiB
Python
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def read_json(path):
|
|
return json.loads(Path(path).read_text(encoding="utf-8"))
|
|
|
|
|
|
def load_models(root):
|
|
return [read_json(path) for path in sorted((root / "models").glob("*.model.json"))]
|
|
|
|
|
|
def load_selector_rules(root):
|
|
return read_json(root / "selector" / "selector_rules.json")
|
|
|
|
|
|
def rules_by_model(selector_rules):
|
|
return {item["model_id"]: item for item in selector_rules.get("models", [])}
|
|
|
|
|
|
def hit_any(user_input, signals):
|
|
return [signal for signal in signals if signal in user_input]
|
|
|
|
|
|
def has_analysis_override(user_input, global_rules):
|
|
return bool(hit_any(user_input, global_rules.get("explicit_analysis_override_phrases", [])))
|
|
|
|
|
|
def hard_no_call_hits(user_input, global_rules):
|
|
signals = list(global_rules.get("hard_no_call_signals", []))
|
|
signals.extend(global_rules.get("direct_execution_no_call_signals", []))
|
|
return hit_any(user_input, signals)
|
|
|
|
|
|
def model_hard_exclusion_hits(user_input, model_rule):
|
|
return hit_any(user_input, model_rule.get("hard_exclusion_triggers", []))
|
|
|
|
|
|
def score_model(model, model_rule, global_rules, user_input, task_type="", pipeline_position=""):
|
|
model_id = model.get("model_id")
|
|
weights = global_rules.get("weights", {})
|
|
score = float(model_rule.get("base_score", 0)) / 100
|
|
reasons = []
|
|
penalties = []
|
|
|
|
negative_hits = hit_any(user_input, model_rule.get("negative_triggers", [])) + hit_any(user_input, model.get("negative_triggers", []))
|
|
if negative_hits and not has_analysis_override(user_input, global_rules):
|
|
penalty_key = "model_negative_penalty_qpi" if model_id == "qpi" else "model_negative_penalty_default"
|
|
score -= weights.get(penalty_key, 0.5)
|
|
penalties.append("negative trigger: " + "、".join(sorted(set(negative_hits))[:4]))
|
|
|
|
trigger_hits = hit_any(user_input, model_rule.get("trigger_keywords", [])) + hit_any(user_input, model.get("trigger_keywords", []))
|
|
trigger_hits = sorted(set(trigger_hits), key=trigger_hits.index)
|
|
if trigger_hits:
|
|
score += weights.get("trigger_keyword", 0.25)
|
|
reasons.append("trigger keyword matched: " + "、".join(trigger_hits[:4]))
|
|
|
|
input_type_matches = model_rule.get("input_type_matches", []) + model.get("input_types", [])
|
|
input_hits = [input_type for input_type in input_type_matches if input_type in user_input or input_type == task_type]
|
|
if input_hits:
|
|
score += weights.get("input_type", 0.15)
|
|
reasons.append("input type matched: " + "、".join(input_hits[:3]))
|
|
|
|
if pipeline_position and model.get("pipeline_position") == pipeline_position:
|
|
score += weights.get("pipeline_position", 0.2)
|
|
reasons.append("pipeline_position matched")
|
|
|
|
complexity_hits = hit_any(user_input, global_rules.get("complexity_signals", []))
|
|
if complexity_hits:
|
|
score += weights.get("complexity_signal", 0.15)
|
|
reasons.append("complexity signal: " + "、".join(complexity_hits[:4]))
|
|
|
|
priority = model.get("selection_priority")
|
|
if isinstance(priority, int):
|
|
score += max(0, min(priority, 10)) * weights.get("selection_priority_factor", 0.01)
|
|
reasons.append(f"selection_priority={priority}")
|
|
|
|
qpi_gate_hits = hit_any(user_input, global_rules.get("qpi_gate_signals", []))
|
|
if model_id == "qpi" and qpi_gate_hits:
|
|
score += weights.get("qpi_gate_bonus", 0.25)
|
|
reasons.append("QPI problem-definition gate matched")
|
|
if model_id == "qpi":
|
|
has_positive_signal = bool(
|
|
trigger_hits
|
|
or input_hits
|
|
or complexity_hits
|
|
or qpi_gate_hits
|
|
or has_analysis_override(user_input, global_rules)
|
|
or task_type in {"problem_definition", "question_analysis"}
|
|
or pipeline_position in {"pre_analysis", "problem_definition"}
|
|
)
|
|
if not has_positive_signal:
|
|
score -= weights.get("qpi_missing_positive_signal_penalty", 0.25)
|
|
penalties.append("QPI positive signal missing")
|
|
if model_id == "qpi" and "QPI 已判断" in user_input:
|
|
score -= weights.get("qpi_already_completed_penalty", 0.25)
|
|
penalties.append("QPI already completed")
|
|
|
|
if model_id == "intellectual_archaeology":
|
|
heavy_hits = hit_any(user_input, global_rules.get("ia_heavy_signals", []))
|
|
if heavy_hits:
|
|
score += weights.get("ia_heavy_bonus", 0.2)
|
|
reasons.append("IA heavy-depth signal: " + "、".join(heavy_hits[:3]))
|
|
if "QPI 已判断" in user_input:
|
|
score += weights.get("ia_qpi_completed_bonus", 0.25)
|
|
reasons.append("QPI-before-IA gate satisfied")
|
|
if global_rules.get("qpi_before_ia_gate", True) and qpi_gate_hits and "QPI 已判断" not in user_input:
|
|
score -= weights.get("ia_qpi_gate_penalty", 0.45)
|
|
penalties.append("QPI-before-IA gate: " + "、".join(qpi_gate_hits[:3]))
|
|
if not heavy_hits and "QPI 已判断" not in user_input:
|
|
score -= weights.get("ia_no_heavy_gate_penalty", 0.2)
|
|
penalties.append("no IA heavy-depth gate")
|
|
|
|
score = min(max(score, 0.0), 1.0)
|
|
return {
|
|
"model_id": model_id,
|
|
"score": round(score, 2),
|
|
"reasons": reasons,
|
|
"penalties": penalties
|
|
}
|
|
|
|
|
|
def hard_no_call_result(root, user_input, hits):
|
|
rejected = []
|
|
for model in load_models(root):
|
|
rejected.append({
|
|
"model_id": model.get("model_id"),
|
|
"score": 0.0,
|
|
"reasons": [],
|
|
"penalties": ["hard no-call signal: " + "、".join(sorted(set(hits))[:4])],
|
|
"decision": "rejected"
|
|
})
|
|
return {
|
|
"input": user_input,
|
|
"selected_models": [],
|
|
"rejected_models": rejected,
|
|
"no_call": True,
|
|
"routing_notes": "hard no-call gate matched; explicit analysis override was not present."
|
|
}
|
|
|
|
|
|
def recommend(root, user_input, task_type="", pipeline_position="", threshold=None):
|
|
selector_rules = load_selector_rules(root)
|
|
global_rules = selector_rules.get("global_rules", {})
|
|
threshold = global_rules.get("no_call_threshold", 0.35) if threshold is None else threshold
|
|
|
|
no_call_hits = hard_no_call_hits(user_input, global_rules)
|
|
if global_rules.get("hard_no_call_first", True) and no_call_hits and not has_analysis_override(user_input, global_rules):
|
|
return hard_no_call_result(root, user_input, no_call_hits)
|
|
|
|
model_rules = rules_by_model(selector_rules)
|
|
scored = []
|
|
for model in load_models(root):
|
|
model_id = model.get("model_id")
|
|
model_rule = model_rules.get(model_id, {})
|
|
item = score_model(model, model_rule, global_rules, user_input, task_type, pipeline_position)
|
|
exclusion_hits = model_hard_exclusion_hits(user_input, model_rule)
|
|
if exclusion_hits:
|
|
item["score"] = 0.0
|
|
item["hard_excluded"] = True
|
|
item["penalties"].append("model hard exclusion signal: " + "、".join(sorted(set(exclusion_hits))[:4]))
|
|
scored.append(item)
|
|
no_call = all(item["score"] < threshold for item in scored)
|
|
selected = []
|
|
rejected = []
|
|
|
|
for item in scored:
|
|
output_item = {
|
|
"model_id": item["model_id"],
|
|
"score": item["score"],
|
|
"reasons": item["reasons"],
|
|
"penalties": item["penalties"],
|
|
"decision": "selected" if item["score"] >= threshold and not no_call and not item.get("hard_excluded") else "rejected"
|
|
}
|
|
if output_item["decision"] == "selected":
|
|
selected.append(output_item)
|
|
else:
|
|
rejected.append(output_item)
|
|
|
|
selected.sort(key=lambda item: item["score"], reverse=True)
|
|
rejected.sort(key=lambda item: item["score"], reverse=True)
|
|
|
|
return {
|
|
"input": user_input,
|
|
"selected_models": selected,
|
|
"rejected_models": rejected,
|
|
"no_call": no_call,
|
|
"routing_notes": "selector is rule-based and driven by selector/selector_rules.json; no LLM, no vector search, no answer generation."
|
|
}
|
|
|
|
|
|
def main():
|
|
root = Path(__file__).resolve().parents[1]
|
|
example = "团队每次都说要长期主义，但一到季度 KPI 就回到短期动作，这到底是什么问题？"
|
|
result = recommend(root, example, task_type="question_analysis", pipeline_position="pre_analysis")
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|