budget.py 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152
  1. #!/usr/bin/env python3
  2. # -*- coding: utf-8 -*-
  3. """
  4. 长期记忆预算分配。
  5. """
  6. from __future__ import annotations
  7. from typing import Dict
  8. DEFAULT_BUDGET = {
  9. "write": {"max_items": 30, "working_ratio": 0.45, "episodic_ratio": 0.30, "semantic_ratio": 0.25},
  10. "review": {"max_items": 40, "working_ratio": 0.35, "episodic_ratio": 0.35, "semantic_ratio": 0.30},
  11. "query": {"max_items": 25, "working_ratio": 0.30, "episodic_ratio": 0.45, "semantic_ratio": 0.25},
  12. }
  13. def get_budget(task_type: str = "write") -> Dict[str, float]:
  14. key = str(task_type or "write").lower()
  15. return dict(DEFAULT_BUDGET.get(key, DEFAULT_BUDGET["write"]))
  16. def allocate_limits(max_items: int, task_type: str = "write") -> Dict[str, int]:
  17. """按任务类型分配 working/episodic/semantic 的条目预算。"""
  18. max_items = max(1, int(max_items or 1))
  19. budget = get_budget(task_type)
  20. wr = float(budget.get("working_ratio", 0.45))
  21. er = float(budget.get("episodic_ratio", 0.30))
  22. sr = float(budget.get("semantic_ratio", 0.25))
  23. total_ratio = wr + er + sr
  24. if total_ratio <= 0:
  25. wr, er, sr = 0.45, 0.30, 0.25
  26. total_ratio = 1.0
  27. wr, er, sr = wr / total_ratio, er / total_ratio, sr / total_ratio
  28. w = int(max_items * wr)
  29. e = int(max_items * er)
  30. s = int(max_items * sr)
  31. used = w + e + s
  32. # 把余数按语义层优先分配,保证总和等于 max_items。
  33. while used < max_items:
  34. if s <= w:
  35. s += 1
  36. elif w <= e:
  37. w += 1
  38. else:
  39. e += 1
  40. used += 1
  41. return {"working": w, "episodic": e, "semantic": s}