Summarize dropped messages using the 0.5B on port 8081, or OpenRouter when CODEY_BACKEND=openrouter. Returns the summary string or None.
(dropped_msgs: list[dict])
| 79 | # ── 0.5B micro-summary ──────────────────────────────────────────────────────── |
| 80 | |
| 81 | def _call_05b(dropped_msgs: list[dict]) -> str | None: |
| 82 | """ |
| 83 | Summarize dropped messages using the 0.5B on port 8081, or OpenRouter |
| 84 | when CODEY_BACKEND=openrouter. Returns the summary string or None. |
| 85 | """ |
| 86 | if not dropped_msgs: |
| 87 | return None |
| 88 | |
| 89 | history_text = "\n".join( |
| 90 | f"{m['role'].upper()}: {m.get('content', '')[:MICRO_SUMMARY_MSG_LIMIT]}" |
| 91 | for m in dropped_msgs |
| 92 | ) |
| 93 | |
| 94 | messages = [ |
| 95 | {"role": "system", "content": _MICRO_SUMMARY_SYSTEM}, |
| 96 | {"role": "user", "content": f"Conversation:\n{history_text}"}, |
| 97 | ] |
| 98 | |
| 99 | # Route to remote planner backend when active — avoids needing the local 0.5B server |
| 100 | try: |
| 101 | from utils.config import is_remote_planner_backend, CODEY_PLANNER_BACKEND |
| 102 | if is_remote_planner_backend(): |
| 103 | from core.inference_openrouter import get_remote_backend |
| 104 | backend = get_remote_backend(CODEY_PLANNER_BACKEND) |
| 105 | result = backend.infer(messages, max_tokens=160, stream=False) |
| 106 | if result: |
| 107 | text, _, _ = result |
| 108 | return text if text else None |
| 109 | return None |
| 110 | except Exception as e: |
| 111 | warning(f"[summarizer] remote micro-summary failed: {e}") |
| 112 | return None |
| 113 | |
| 114 | # Local 0.5B path |
| 115 | payload = { |
| 116 | "model": "codey-planner", |
| 117 | "messages": messages, |
| 118 | "max_tokens": 160, |
| 119 | "temperature": 0.2, |
| 120 | "stream": False, |
| 121 | } |
| 122 | |
| 123 | url = f"http://{_05B_HOST}:{_05B_PORT}/v1/chat/completions" |
| 124 | req = urllib.request.Request( |
| 125 | url, |
| 126 | data=json.dumps(payload).encode("utf-8"), |
| 127 | headers={"Content-Type": "application/json"}, |
| 128 | method="POST", |
| 129 | ) |
| 130 | |
| 131 | try: |
| 132 | with urllib.request.urlopen(req, timeout=30) as resp: |
| 133 | result = json.loads(resp.read().decode("utf-8")) |
| 134 | choices = result.get("choices", []) |
| 135 | if choices: |
| 136 | text = choices[0].get("message", {}).get("content", "").strip() |
| 137 | return text if text else None |
| 138 | except Exception as e: |
no test coverage detected