| | """ |
| | Single Planner Module |
| | |
| | A single-LLM-call planner for ablation comparison with Council-Lite. |
| | Uses the same prompt template and schema as UnifiedPlanner but |
| | provides explicit token tracking for fair comparison. |
| | """ |
| |
|
| | from __future__ import annotations |
| |
|
| | from dataclasses import dataclass |
| | from pathlib import Path |
| | from typing import Any, Dict, Optional |
| |
|
| | from src.planner.ollama_llm import OllamaPlannerLLM |
| | from src.planner.schema import SemanticPlan |
| | from src.planner.validation import validate_semantic_plan_dict |
| |
|
| |
|
| | @dataclass |
| | class PlannerMetrics: |
| | """Metrics for tracking planner resource usage.""" |
| | input_tokens: int = 0 |
| | output_tokens: int = 0 |
| | total_tokens: int = 0 |
| | llm_calls: int = 0 |
| | generation_time_ms: float = 0.0 |
| |
|
| | def to_dict(self) -> Dict[str, Any]: |
| | """Convert to dictionary.""" |
| | return { |
| | "input_tokens": self.input_tokens, |
| | "output_tokens": self.output_tokens, |
| | "total_tokens": self.total_tokens, |
| | "llm_calls": self.llm_calls, |
| | "generation_time_ms": self.generation_time_ms, |
| | } |
| |
|
| |
|
| | class SinglePlannerLLM: |
| | """ |
| | Single-call semantic planner. |
| | |
| | This is the baseline planner condition for ablation: |
| | - 1 LLM call |
| | - Standard token budget |
| | - Full semantic plan output |
| | |
| | Equivalent to UnifiedPlanner but with explicit metrics. |
| | """ |
| |
|
| | name = "SinglePlanner" |
| |
|
| | def __init__( |
| | self, |
| | prompt_path: str = "src/planner/prompts/unified.txt", |
| | model: str = "qwen2:7b", |
| | base_url: str = "http://localhost:11434", |
| | ): |
| | self.llm = OllamaPlannerLLM(model=model, base_url=base_url) |
| | self.prompt_template = Path(prompt_path).read_text(encoding="utf-8") |
| | self.metrics: Optional[PlannerMetrics] = None |
| |
|
| | def plan(self, user_prompt: str) -> SemanticPlan: |
| | """ |
| | Generate a semantic plan from user prompt. |
| | |
| | Args: |
| | user_prompt: The user's scene description |
| | |
| | Returns: |
| | SemanticPlan object |
| | """ |
| | import time |
| |
|
| | |
| | prompt = self.prompt_template.replace("{{USER_PROMPT}}", user_prompt) |
| |
|
| | |
| | start_time = time.time() |
| |
|
| | |
| | data = self.llm.generate_json(prompt) |
| |
|
| | end_time = time.time() |
| |
|
| | |
| | input_tokens = len(prompt.split()) * 1.3 |
| | output_tokens = len(str(data).split()) * 1.3 |
| |
|
| | self.metrics = PlannerMetrics( |
| | input_tokens=int(input_tokens), |
| | output_tokens=int(output_tokens), |
| | total_tokens=int(input_tokens + output_tokens), |
| | llm_calls=1, |
| | generation_time_ms=(end_time - start_time) * 1000, |
| | ) |
| |
|
| | |
| | validate_semantic_plan_dict(data) |
| | return SemanticPlan(**data) |
| |
|
| | def get_metrics(self) -> Optional[PlannerMetrics]: |
| | """Get metrics from the last plan() call.""" |
| | return self.metrics |
| |
|
| |
|
| | |
| | SinglePlanner = SinglePlannerLLM |
| |
|