Spaces:
Running
Running
File size: 2,542 Bytes
3040bf7 | 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 | """ProgramLoader — reads apishift_program.md and selects the section
relevant to the current step.
The Manager's observation includes a 'program_section' field. We don't
inject the entire manual every step (context bloat). Instead we pick the
section that most directly applies to the most recent action context.
"""
import os
import re
from typing import Dict, List
SECTION_RE = re.compile(r"^##\s+\d+\.\s+(.*?)\s*$", re.MULTILINE)
class ProgramLoader:
def __init__(self, program_path: str = "apishift_program.md"):
self.program_path = program_path
self._sections: Dict[str, str] = {}
self._load()
def _load(self):
if not os.path.exists(self.program_path):
return
with open(self.program_path, "r", encoding="utf-8") as f:
text = f.read()
# Split into sections at "## N. <title>" headers
parts = re.split(r"(?m)^##\s+(\d+\.\s+.*?)$", text)
# parts looks like: [intro, 'header_1', 'body_1', 'header_2', 'body_2', ...]
for i in range(1, len(parts) - 1, 2):
header = parts[i].strip()
body = parts[i + 1].strip()
self._sections[header] = body
def section_for_step(self, last_command: str = "", steps_taken: int = 0) -> str:
"""Return the most relevant section text for the current step.
Falls back to the setup ritual section at the start of an episode."""
if not self._sections:
return ""
keywords = {
"inspect": "1. Setup ritual",
"dispatch_diff": "2. Action ordering rules",
"classify_impact": "2. Action ordering rules",
"dispatch_patch": "2. Action ordering rules",
"dispatch_test": "4. Failure handling",
"dispatch_rollback": "2. Action ordering rules",
"read_memory": "5. Memory usage rules",
"submit": "7. Step budget management",
}
if steps_taken == 0:
target = "1. Setup ritual"
else:
target = keywords.get(last_command, "3. Simplicity criterion")
for header, body in self._sections.items():
if header.startswith(target.split(".")[0]):
return f"## {header}\n{body}"
# Fallback to first section
first = next(iter(self._sections.items()))
return f"## {first[0]}\n{first[1]}"
def all_sections(self) -> List[str]:
return [f"## {h}\n{b}" for h, b in self._sections.items()]
|