243 lines
9.8 KiB
Python
243 lines
9.8 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 instruction_segment(user_input):
|
||
segments = [user_input]
|
||
for delimiter in ("：", ":", "\n"):
|
||
if delimiter in user_input:
|
||
segments.append(user_input.split(delimiter, 1)[0])
|
||
return min(segments, key=len)
|
||
|
||
|
||
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", []))
|
||
hits = hit_any(user_input, signals)
|
||
instruction_hits = instruction_task_no_call_hits(
|
||
user_input,
|
||
global_rules.get("instruction_task_no_call_signals", []),
|
||
)
|
||
return hits + instruction_hits
|
||
|
||
|
||
def instruction_task_no_call_hits(user_input, signals):
|
||
instruction = instruction_segment(user_input)
|
||
if instruction != user_input:
|
||
return hit_any(instruction, signals)
|
||
stripped = user_input.strip()
|
||
return [
|
||
signal
|
||
for signal in signals
|
||
if 0 <= stripped.find(signal) <= 4
|
||
]
|
||
|
||
|
||
def depth_limited_qpi_override_hits(user_input, global_rules):
|
||
depth_hits = hit_any(user_input, global_rules.get("depth_limiting_no_call_signals", []))
|
||
override_hits = hit_any(user_input, global_rules.get("qpi_limited_analysis_override_phrases", []))
|
||
return depth_hits, override_hits
|
||
|
||
|
||
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", []))
|
||
depth_hits, qpi_limited_hits = depth_limited_qpi_override_hits(user_input, global_rules)
|
||
suppress_qpi_depth_penalty = model_id == "qpi" and depth_hits and qpi_limited_hits
|
||
if negative_hits and not has_analysis_override(user_input, global_rules) and not suppress_qpi_depth_penalty:
|
||
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)
|
||
depth_hits, qpi_limited_hits = depth_limited_qpi_override_hits(user_input, global_rules)
|
||
has_depth_limited_qpi_override = bool(depth_hits and qpi_limited_hits)
|
||
if (
|
||
global_rules.get("hard_no_call_first", True)
|
||
and no_call_hits
|
||
and not has_analysis_override(user_input, global_rules)
|
||
and not has_depth_limited_qpi_override
|
||
):
|
||
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())
|