File size: 4,564 Bytes
f440f03 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """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]
|