| """Autonomous planning heuristics used by the Python runtime.""" |
|
|
| from __future__ import annotations |
|
|
| from typing import Any |
|
|
| from maris_core.memory_context import MemoryMatch |
|
|
| _CODE_KEYWORDS = ("kod", "api", "script", "python", "rust", "refactor", "fix") |
| _BROWSER_KEYWORDS = ("browser", "web", "pārlūk", "klikš", "scrape", "form", "http://", "https://") |
| _VALIDATION_KEYWORDS = ("test", "verify", "pārbaud", "validate", "review") |
| _RESEARCH_KEYWORDS = ("research", "meklē", "salīdzini", "izpēti") |
| _PUNCTUATION_SEPARATORS = {",", ";", "->"} |
| _WORD_SEPARATORS = {"and", "un", "then"} |
|
|
|
|
| class Planner: |
| """Produces structured task graphs with lightweight policy hints.""" |
|
|
| def describe(self, goal: str, max_steps: int = 10) -> list[str]: |
| structured = self.decompose(goal, max_steps=max_steps) |
| if structured: |
| return [str(step["action"]) for step in structured] |
| return [ |
| f"Izanalizēt mērķi un izpildes robežas: {goal}", |
| "Izpildīt galveno darba soli ar atbilstošu rīku", |
| "Validēt rezultātu un sagatavot nākamos soļus", |
| ][: max(1, max_steps)] |
|
|
| def decompose( |
| self, |
| goal: str, |
| max_steps: int = 10, |
| memory_context: list[MemoryMatch] | None = None, |
| ) -> list[dict[str, Any]]: |
| normalized_goal = goal.strip() |
| if not normalized_goal: |
| return [] |
|
|
| chunks = _split_goal_chunks(normalized_goal) |
| candidate_steps = chunks[: max_steps - 1] if chunks else [] |
| if not candidate_steps: |
| candidate_steps.append(normalized_goal) |
|
|
| actions: list[str] = [f"Izanalizēt mērķi un atkarības: {normalized_goal}"] |
| actions.extend(candidate_steps[: max(0, max_steps - 2)]) |
| if len(actions) < max_steps: |
| actions.append("Pārbaudīt rezultātu, riskus un resumējamu stāvokli") |
|
|
| if memory_context: |
| actions.insert( |
| 1, |
| "Ielādēt un izmantot atbilstošo sesijas/memory kontekstu pirms galvenās izpildes", |
| ) |
|
|
| plan: list[dict[str, Any]] = [] |
| for index, action in enumerate(actions[:max_steps], start=1): |
| tool = self._infer_tool(action) |
| risk_level = "high" if tool in {"browser_automation", "code_generation"} else "medium" |
| plan.append( |
| { |
| "step": index, |
| "action": action, |
| "tool": tool, |
| "depends_on_steps": [index - 1] if index > 1 else [], |
| "execution_policy": "sequential", |
| "risk_level": risk_level, |
| "approval_required": tool |
| in {"browser_automation", "code_generation", "validation"}, |
| "max_attempts": 1 if tool == "validation" else 2, |
| "observability_tags": ["autonomous", tool, f"risk:{risk_level}"], |
| } |
| ) |
| return plan |
|
|
| def _infer_tool(self, action: str) -> str: |
| lowered = action.lower() |
| if any(keyword in lowered for keyword in _BROWSER_KEYWORDS): |
| return "browser_automation" |
| if any(keyword in lowered for keyword in _CODE_KEYWORDS): |
| return "code_generation" |
| if any(keyword in lowered for keyword in _VALIDATION_KEYWORDS): |
| return "validation" |
| if any(keyword in lowered for keyword in _RESEARCH_KEYWORDS): |
| return "web_research" |
| return "reasoning" |
|
|
|
|
| def _split_goal_chunks(goal: str) -> list[str]: |
| normalized = goal.replace("->", " -> ").replace(",", " , ").replace(";", " ; ") |
| tokens = normalized.split() |
| chunks: list[str] = [] |
| current_tokens: list[str] = [] |
| index = 0 |
|
|
| while index < len(tokens): |
| token = tokens[index] |
| lowered = token.lower() |
| next_lowered = tokens[index + 1].lower() if index + 1 < len(tokens) else "" |
| is_separator = ( |
| token in _PUNCTUATION_SEPARATORS |
| or lowered in _WORD_SEPARATORS |
| or (lowered == "un" and next_lowered == "tad") |
| ) |
| if is_separator: |
| if current_tokens: |
| chunks.append(" ".join(current_tokens).strip(" -")) |
| current_tokens = [] |
| index += 2 if lowered == "un" and next_lowered == "tad" else 1 |
| continue |
|
|
| current_tokens.append(token) |
| index += 1 |
|
|
| if current_tokens: |
| chunks.append(" ".join(current_tokens).strip(" -")) |
|
|
| return [chunk for chunk in chunks if chunk] |
|
|
