Files
manual_slop/orchestrator_pm.py
2026-02-26 22:06:18 -05:00

66 lines
2.1 KiB
Python

import json
import ai_client
import mma_prompts
import aggregate
import summarize
from pathlib import Path
def generate_tracks(user_request: str, project_config: dict, file_items: list[dict]) -> list[dict]:
"""
Tier 1 (Strategic PM) call.
Analyzes the project state and user request to generate a list of Tracks.
"""
# 1. Build Repository Map (Summary View)
repo_map = summarize.build_summary_markdown(file_items)
# 2. Construct Prompt
system_prompt = mma_prompts.PROMPTS.get("tier1_epic_init")
user_message = (
f"### USER REQUEST:
{user_request}
"
f"### REPOSITORY MAP:
{repo_map}
"
"Please generate the implementation tracks for this request."
)
# 3. Call Tier 1 Model (Strategic - Pro)
# Note: We use gemini-1.5-pro or similar high-reasoning model for Tier 1
response = ai_client.send(
md_content="", # We pass everything in user_message for clarity
user_message=user_message,
system_prompt=system_prompt,
model_name="gemini-1.5-pro" # Strategic Tier
)
# 4. Parse JSON Output
try:
# The prompt asks for a JSON array. We need to extract it if the AI added markdown blocks.
json_match = response.strip()
if "```json" in json_match:
json_match = json_match.split("```json")[1].split("```")[0].strip()
elif "```" in json_match:
json_match = json_match.split("```")[1].split("```")[0].strip()
tracks = json.loads(json_match)
return tracks
except Exception as e:
print(f"Error parsing Tier 1 response: {e}")
print(f"Raw response: {response}")
return []
if __name__ == "__main__":
# Quick CLI test
import project_manager
proj = project_manager.load_project("manual_slop.toml")
flat = project_manager.flat_config(proj)
file_items = aggregate.build_file_items(Path("."), flat.get("files", {}).get("paths", []))
print("Testing Tier 1 Track Generation...")
tracks = generate_tracks("Implement a basic unit test for the ai_client.py module.", flat, file_items)
print(json.dumps(tracks, indent=2))