Route planning through the active planner backend (OpenRouter or UnlimitedClaude).
(prompt: str)
| 136 | # ── Planning via 0.5B on port 8081 (or remote when CODEY_BACKEND_P is set) ── |
| 137 | |
| 138 | def _get_plan_remote(prompt: str) -> Optional[List[str]]: |
| 139 | """Route planning through the active planner backend (OpenRouter or UnlimitedClaude).""" |
| 140 | try: |
| 141 | from utils.config import ( |
| 142 | PLANNER_TEMPERATURE, PLANNER_MAX_TOKENS, CODEY_PLANNER_BACKEND, |
| 143 | OPENROUTER_PLANNER_MODEL, OPENROUTER_BASE_URL, OPENROUTER_API_KEY, |
| 144 | UNLIMITEDCLAUDE_PLANNER_MODEL, UNLIMITEDCLAUDE_BASE_URL, UNLIMITEDCLAUDE_API_KEY, |
| 145 | ) |
| 146 | from utils.logger import info, warning |
| 147 | |
| 148 | if CODEY_PLANNER_BACKEND == "unlimitedclaude": |
| 149 | planner_model = UNLIMITEDCLAUDE_PLANNER_MODEL |
| 150 | base_url = UNLIMITEDCLAUDE_BASE_URL.rstrip("/") |
| 151 | api_key = UNLIMITEDCLAUDE_API_KEY |
| 152 | backend_label = "unlimitedclaude" |
| 153 | else: |
| 154 | planner_model = OPENROUTER_PLANNER_MODEL |
| 155 | base_url = OPENROUTER_BASE_URL.rstrip("/") |
| 156 | api_key = OPENROUTER_API_KEY |
| 157 | backend_label = "openrouter" |
| 158 | |
| 159 | messages = [ |
| 160 | {"role": "system", "content": PLANNER_PROMPT}, |
| 161 | {"role": "user", "content": prompt}, |
| 162 | ] |
| 163 | |
| 164 | # Use the dedicated planner model and low temperature (0.2 not 0.7) |
| 165 | import json as _json |
| 166 | import urllib.request as _req |
| 167 | payload = { |
| 168 | "model": planner_model, |
| 169 | "messages": messages, |
| 170 | "max_tokens": PLANNER_MAX_TOKENS, |
| 171 | "temperature": PLANNER_TEMPERATURE, |
| 172 | "stream": False, |
| 173 | } |
| 174 | headers = { |
| 175 | "Content-Type": "application/json", |
| 176 | "Authorization": f"Bearer {api_key}", |
| 177 | "HTTP-Referer": "https://github.com/codey-v2", |
| 178 | "X-Title": "Codey-v2", |
| 179 | } |
| 180 | request = _req.Request( |
| 181 | f"{base_url}/chat/completions", |
| 182 | data=_json.dumps(payload).encode("utf-8"), |
| 183 | headers=headers, |
| 184 | method="POST", |
| 185 | ) |
| 186 | try: |
| 187 | with _req.urlopen(request, timeout=60) as resp: |
| 188 | result = _json.loads(resp.read().decode("utf-8")) |
| 189 | msg = result["choices"][0].get("message", {}) |
| 190 | # content can be null when the model returns a tool_call instead of text |
| 191 | content = msg.get("content") or "" |
| 192 | # Qwen3 / thinking models put output in reasoning_content when content is empty |
| 193 | if not content: |
| 194 | content = msg.get("reasoning_content") or "" |
| 195 | # some models return text inside tool_calls[0].function.arguments |
no test coverage detected