Function get_plan

core/plannd.py:223–285

Break *prompt* into a numbered plan. Uses the local 0.5B on port 8081 by default. When CODEY_BACKEND_P (or CODEY_BACKEND) is a remote backend, routes there instead so the 0.5B server does not need to be running.

(prompt: str)

def get_plan(prompt: str) -> Optional[List[str]]:
    """
    Break *prompt* into a numbered plan.

    Uses the local 0.5B on port 8081 by default.
    When CODEY_BACKEND_P (or CODEY_BACKEND) is a remote backend, routes
    there instead so the 0.5B server does not need to be running.
    """
    try:
        from utils.config import is_remote_planner_backend
        if is_remote_planner_backend():
            return _get_plan_remote(prompt)
    except ImportError:
        pass

    try:
        from utils.config import PLANNER_TEMPERATURE, PLANNER_MAX_TOKENS
        temperature = PLANNER_TEMPERATURE
        max_tokens  = PLANNER_MAX_TOKENS
    except ImportError:
        temperature = 0.2
        max_tokens  = 512

    try:
        from utils.config import PLANND_SERVER_PORT
        port = PLANND_SERVER_PORT
    except ImportError:
        port = 8081

    payload = {
        "model": "plannd",
        "messages": [
            {"role": "system", "content": PLANNER_PROMPT},
            {"role": "user",   "content": prompt},
        ],
        "max_tokens": max_tokens,
        "temperature": temperature,
        "stream": False,
    }

    url = f"http://127.0.0.1:{port}/v1/chat/completions"
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, timeout=60) as response:
            result = json.loads(response.read().decode("utf-8"))
            choices = result.get("choices", [])
            if not choices:
                return None
            raw = choices[0].get("message", {}).get("content", "").strip()
            if not raw:
                return None
            steps = parse_steps(raw)