OUTPUT #691
PARÇA 4 / 5
TOPLAM: 167636 karakter | 2901 satır
BU PARÇA: 40000 karakter
--
90- f.write(json.dumps(entry, ensure_ascii=False) + '\n')
91-
92-
93-# YAML section -> keys; "yaml_key:attr" when the config attribute name differs.
94-_YAML_SECTIONS: Dict[str, Tuple[str, ...]] = {
95- "tokenizer": ("name:tokenizer_name", "trust_remote_code"),
96: "compression": ("target_max_tokens", "summary_target_tokens"),
97- "protected_turns": ("first_system:protect_first_system", "first_human:protect_first_human",
98- "first_gpt:protect_first_gpt", "first_tool:protect_first_tool", "last_n_turns:protect_last_n_turns"),
99: "summarization": ("model:summarization_model", "base_url", "api_key_env", "temperature", "max_retries", "retry_delay"),
100- "output": ("add_summary_notice", "summary_notice_text", "output_suffix"),
101: "processing": ("num_workers", "max_concurrent_requests", "skip_under_target", "save_over_limit"),
102- "metrics": ("enabled:metrics_enabled", "per_trajectory:metrics_per_trajectory", "output_file:metrics_output_file"),
103-}
104-
105-
106-@dataclass
107:class CompressionConfig:
108: """Configuration for trajectory compression (tokenizer / targets / protected turns / summarizer / output / processing / metrics)."""
109- tokenizer_name: str = "moonshotai/Kimi-K2-Thinking"
110- trust_remote_code: bool = True
111- target_max_tokens: int = 15250
112- summary_target_tokens: int = 750
113- protect_first_system: bool = True
114- protect_first_human: bool = True
115- protect_first_gpt: bool = True
116- protect_first_tool: bool = True
117- protect_last_n_turns: int = 4
118: summarization_model: str = "google/gemini-3-flash-preview"
119- base_url: str = OPENROUTER_BASE_URL
120- api_key_env: str = "OPENROUTER_API_KEY"
121- temperature: float = 0.3
122- max_retries: int = 3
123: retry_delay: int = 2
124- add_summary_notice: bool = True
125: summary_notice_text: str = "\n\nSome of your previous tool responses may be summarized to preserve context."
126: output_suffix: str = "_compressed"
127- num_workers: int = 4
128- max_concurrent_requests: int = 50 # Max concurrent API calls for summarization
129- skip_under_target: bool = True
130: save_over_limit: bool = True
131: per_trajectory_timeout: int = 300 # seconds (default: 5 min)
132- metrics_enabled: bool = True
133- metrics_per_trajectory: bool = True
134: metrics_output_file: str = "compression_metrics.json"
135-
136- @classmethod
137: def from_yaml(cls, yaml_path: str) -> "CompressionConfig":
138- """Load configuration from YAML file (missing keys keep the defaults)."""
139- with open(yaml_path, 'r', encoding="utf-8") as f:
140- data = yaml.safe_load(f) or {}
141- config = cls()
142- for section, keys in _YAML_SECTIONS.items():
143- for key in keys if section in data else ():
--
149- setattr(config, attr, value)
150- return config
151-
152-
153-@dataclass
154-class TrajectoryMetrics:
155: """Metrics for a single trajectory compression."""
156- original_tokens: int = 0
157: compressed_tokens: int = 0
158- tokens_saved: int = 0
159: compression_ratio: float = 1.0
160- original_turns: int = 0
161: compressed_turns: int = 0
162- turns_removed: int = 0
163: turns_compressed_start_idx: int = -1
164: turns_compressed_end_idx: int = -1
165: turns_in_compressed_region: int = 0
166: was_compressed: bool = False
167: still_over_limit: bool = False
168- skipped_under_target: bool = False
169- summarization_api_calls: int = 0
170- summarization_errors: int = 0
171-
172- def to_dict(self) -> Dict[str, Any]:
173- d = asdict(self)
174: d["compression_ratio"] = round(self.compression_ratio, 4)
175: region = {"start_idx": d.pop("turns_compressed_start_idx"), "end_idx": d.pop("turns_compressed_end_idx"),
176: "turns_count": d.pop("turns_in_compressed_region")}
177- items = list(d.items())
178: items.insert(7, ("compression_region", region)) # after turns_removed: historical key order
179- return dict(items)
180-
181-
182-def _mean(values, default):
183- return sum(values) / len(values) if values else default
184-
185-
186-@dataclass
187-class AggregateMetrics:
188- """Aggregate metrics across all trajectories."""
189- total_trajectories: int = 0
190: trajectories_compressed: int = 0
191- trajectories_skipped_under_target: int = 0
192: trajectories_still_over_limit: int = 0
193- trajectories_failed: int = 0
194- total_tokens_before: int = 0
195- total_tokens_after: int = 0
196- total_tokens_saved: int = 0
197- total_turns_before: int = 0
198- total_turns_after: int = 0
199- total_turns_removed: int = 0
200- total_summarization_calls: int = 0
201- total_summarization_errors: int = 0
202: compression_ratios: List[float] = field(default_factory=list)
203- tokens_saved_list: List[int] = field(default_factory=list)
204- turns_removed_list: List[int] = field(default_factory=list)
205- processing_start_time: str = ""
206- processing_end_time: str = ""
207- processing_duration_seconds: float = 0.0
208-
209- def add_trajectory_metrics(self, metrics: TrajectoryMetrics):
210- """Add a trajectory's metrics to the aggregate."""
211- self.total_trajectories += 1
212- self.total_tokens_before += metrics.original_tokens
213: self.total_tokens_after += metrics.compressed_tokens
214- self.total_tokens_saved += metrics.tokens_saved
215- self.total_turns_before += metrics.original_turns
216: self.total_turns_after += metrics.compressed_turns
217- self.total_turns_removed += metrics.turns_removed
218- self.total_summarization_calls += metrics.summarization_api_calls
219- self.total_summarization_errors += metrics.summarization_errors
220: if metrics.was_compressed:
221: self.trajectories_compressed += 1
222: self.compression_ratios.append(metrics.compression_ratio)
223- self.tokens_saved_list.append(metrics.tokens_saved)
224- self.turns_removed_list.append(metrics.turns_removed)
225- self.trajectories_skipped_under_target += bool(metrics.skipped_under_target)
226: self.trajectories_still_over_limit += bool(metrics.still_over_limit)
227-
228- def to_dict(self) -> Dict[str, Any]:
229- return {
230: "summary": {"total_trajectories": self.total_trajectories, "trajectories_compressed": self.trajectories_compressed,
231- "trajectories_skipped_under_target": self.trajectories_skipped_under_target,
232: "trajectories_still_over_limit": self.trajectories_still_over_limit, "trajectories_failed": self.trajectories_failed,
233: "compression_rate": round(self.trajectories_compressed / max(self.total_trajectories, 1), 4)},
234- "tokens": {"total_before": self.total_tokens_before, "total_after": self.total_tokens_after, "total_saved": self.total_tokens_saved,
235: "overall_compression_ratio": round(self.total_tokens_after / max(self.total_tokens_before, 1), 4)},
236- "turns": {"total_before": self.total_turns_before, "total_after": self.total_turns_after, "total_removed": self.total_turns_removed},
237: "averages": {"avg_compression_ratio": round(_mean(self.compression_ratios, 1.0), 4),
238: "avg_tokens_saved_per_compressed": round(_mean(self.tokens_saved_list, 0), 1),
239: "avg_turns_removed_per_compressed": round(_mean(self.turns_removed_list, 0), 2)},
240- "summarization": {"total_api_calls": self.total_summarization_calls, "total_errors": self.total_summarization_errors,
241- "success_rate": round(1 - (self.total_summarization_errors / max(self.total_summarization_calls, 1)), 4)},
242- "processing": {"start_time": self.processing_start_time, "end_time": self.processing_end_time,
243- "duration_seconds": round(self.processing_duration_seconds, 2)},
244- }
245-
--
248-_PROVIDER_HOSTS: Tuple[Tuple[str, str], ...] = (
249- ("openrouter.ai", "openrouter"), ("nousresearch.com", "nous"), ("z.ai", "zai"), ("moonshot.ai", "kimi-coding"),
250- ("moonshot.cn", "kimi-coding"), ("api.kimi.com", "kimi-coding"), ("arcee.ai", "arcee"), ("minimaxi.com", "minimax-cn"),
251- ("minimax.io", "minimax"),
252-)
253-
254:_SUMMARY_FALLBACK = "[CONTEXT SUMMARY]: [Summary generation failed - previous turns contained tool calls and responses that have been compressed to save context space.]"
255:_STATUS_FMT = "[dim]✅ {compressed} compressed | ⏭️ {skipped} skipped | ⏱️ {timeouts} timeout | 🔄 {api_calls} API calls | ⚡ {in_flight} in-flight[/dim]"
256-
257-
258-@dataclass
259-class _RunProgress:
260- """Shared counters + rich progress handles for one directory run."""
261- progress: Any
262- main_task: Any
263- status_task: Any
264- lock: asyncio.Lock
265- semaphore: asyncio.Semaphore
266: compressed: int = 0
267- skipped: int = 0
268- api_calls: int = 0
269- in_flight: int = 0
270: timeouts: int = 0
271-
272: def finish(self, update_status: bool = True) -> None:
273- """Retire one in-flight entry and advance the bar (caller holds ``lock``)."""
274- self.in_flight -= 1
275- self.progress.advance(self.main_task)
276- if update_status:
277- self.progress.update(self.status_task, description=_STATUS_FMT.format(
278: compressed=self.compressed, skipped=self.skipped, timeouts=self.timeouts,
279- api_calls=self.api_calls, in_flight=self.in_flight))
280-
281-
282:class TrajectoryCompressor:
283: """Compresses agent trajectories to fit within a target token budget.
284-
285- Keeps protected head/tail turns, summarizes only as much of the middle as
286- needed into one human summary turn, and keeps the remaining middle intact.
287- """
288-
289: def __init__(self, config: CompressionConfig):
290- self.config = config
291- self.aggregate_metrics = AggregateMetrics()
292- self._init_tokenizer()
293- self._init_summarizer()
294- self.logger = logging.getLogger(__name__)
295-
--
305- def _init_summarizer(self):
306- """Route summarization through call_llm for known providers, else a raw client."""
307- provider = self._detect_provider()
308- self._use_call_llm = bool(provider)
309- if provider:
310- self._llm_provider = provider
311: from agent.auxiliary_client import resolve_provider_client
312: client, _ = resolve_provider_client(provider, model=self.config.summarization_model)
313- if client is None:
314- raise RuntimeError(f"Provider '{provider}' is not configured. Check your API key or run: hermes setup")
315- self.client = self.async_client = None # Not used directly
316- else:
317- # Custom endpoint — use config's raw base_url + api_key_env
318- api_key = os.getenv(self.config.api_key_env)
319- if not api_key:
320- raise RuntimeError(f"Missing API key. Set {self.config.api_key_env} environment variable.")
321- from openai import OpenAI
322: from agent.auxiliary_client import _to_openai_base_url
323- self.client = OpenAI(api_key=api_key, base_url=_to_openai_base_url(self.config.base_url))
324- # AsyncOpenAI is created lazily in _get_async_client() so it binds to the current event
325- # loop — each process_directory() runs its own asyncio.run(); a shared client would hit
326- # "Event loop is closed".
327- self.async_client = None
328- self._async_client_api_key = api_key
329: print(f"✅ Initialized summarizer client: {self.config.summarization_model}")
330- print(f" Max concurrent requests: {self.config.max_concurrent_requests}")
331-
332- def _get_async_client(self):
333- """Return a fresh AsyncOpenAI client bound to the running event loop."""
334- from openai import AsyncOpenAI
335: from agent.auxiliary_client import _to_openai_base_url
336- self.async_client = AsyncOpenAI(api_key=self._async_client_api_key, base_url=_to_openai_base_url(self.config.base_url))
337- return self.async_client
338-
339- def _detect_provider(self) -> str:
340- """Provider name for the configured base_url, or ``""`` when unknown."""
341- url = self.config.base_url or ""
--
356- return sum(self.count_turn_tokens(trajectory))
357-
358- def count_turn_tokens(self, trajectory: List[Dict[str, str]]) -> List[int]:
359- return [self.count_tokens(turn.get("value", "")) for turn in trajectory]
360-
361- def _find_protected_indices(self, trajectory: List[Dict[str, str]]) -> Tuple[set, int, int]:
362: """Return ``(protected_set, compressible_start, compressible_end)``."""
363- n = len(trajectory)
364- first_seen: Dict[str, int] = {}
365- for i, turn in enumerate(trajectory):
366- first_seen.setdefault(turn.get("from", ""), i)
367- protected = {first_seen[role] for role in ("system", "human", "gpt", "tool")
368- if getattr(self.config, f"protect_first_{role}") and role in first_seen}
369- protected.update(range(max(0, n - self.config.protect_last_n_turns), n))
370: # Compressible region: after the last protected head turn, before the first tail turn.
371- head_protected = [i for i in protected if i < n // 2]
372- tail_protected = [i for i in protected if i >= n // 2]
373- return protected, max(head_protected) + 1 if head_protected else 0, min(tail_protected) if tail_protected else n
374-
375- @staticmethod
376- def _snap_boundary(trajectory: List[Dict[str, str]], idx: int, min_idx: int, max_idx: int) -> int:
--
418-
419----
420-TURNS TO SUMMARIZE:
421-{content}
422----
423-
424:Write only the summary, starting with "[CONTEXT SUMMARY]:" prefix."""
425-
426- def _summary_request(self, prompt: str) -> Tuple[Optional[float], Dict[str, Any]]:
427- """Return ``(temperature, create-kwargs)``; temperature None means omit it."""
428- cfg = self.config
429: temperature = _effective_temperature_for_model(cfg.summarization_model, cfg.temperature, cfg.base_url)
430: kwargs = {"model": cfg.summarization_model, "messages": [{"role": "user", "content": prompt}],
431- "max_tokens": cfg.summary_target_tokens * 2}
432- if not getattr(self, '_use_call_llm', False) and temperature is not None:
433- kwargs["temperature"] = temperature
434- return temperature, kwargs
435-
436: def _finish_summary(self, response: Any) -> str:
437: """Extract the summary text with the ``[CONTEXT SUMMARY]:`` prefix exactly once; a ``length`` stop is a failure."""
438: if _response_finish_reason(response) == "length":
439- # Storing a truncated summary silently corrupts the trajectory's memory, so raise and
440: # let the retry/backoff loop handle it.
441: raise RuntimeError("trajectory summarization hit the output token cap (finish_reason=length); summary is incomplete")
442- content = response.choices[0].message.content
443- text = (content if isinstance(content, str) else str(content) if content else "").strip()
444: if text.startswith("[CONTEXT SUMMARY]:"):
445- return text
446: return "[CONTEXT SUMMARY]:" if not text else f"[CONTEXT SUMMARY]: {text}"
447-
448- def _summary_attempt_failed(self, metrics: TrajectoryMetrics, attempt: int, exc: Exception) -> Optional[float]:
449- """Record a failed attempt; return the backoff delay, or None on the last attempt."""
450- metrics.summarization_errors += 1
451- self.logger.warning("Summarization attempt %d failed: %s", attempt + 1, exc)
452- if attempt < self.config.max_retries - 1:
453: return jittered_backoff(attempt + 1, base_delay=self.config.retry_delay, max_delay=30.0)
454- return None
455-
456- def _generate_summary(self, content: str, metrics: TrajectoryMetrics) -> str:
457: """Summarize ``content`` with retries; returns a fallback summary after the last failure."""
458- prompt = self._summary_prompt(content)
459- for attempt in range(self.config.max_retries):
460- try:
461- metrics.summarization_api_calls += 1
462- temperature, kwargs = self._summary_request(prompt)
463- if getattr(self, '_use_call_llm', False):
464: from agent.auxiliary_client import call_llm
465- response = call_llm(provider=self._llm_provider, temperature=temperature, **kwargs)
466- else:
467- response = self.client.chat.completions.create(**kwargs)
468: return self._finish_summary(response)
469- except Exception as e:
470- delay = self._summary_attempt_failed(metrics, attempt, e)
471- if delay is None:
472: return _SUMMARY_FALLBACK
473- time.sleep(delay)
474-
475- async def _generate_summary_async(self, content: str, metrics: TrajectoryMetrics) -> str:
476- """Async twin of ``_generate_summary``."""
477- prompt = self._summary_prompt(content)
478- for attempt in range(self.config.max_retries):
479- try:
480- metrics.summarization_api_calls += 1
481- temperature, kwargs = self._summary_request(prompt)
482- if getattr(self, '_use_call_llm', False):
483: from agent.auxiliary_client import async_call_llm
484- response = await async_call_llm(provider=self._llm_provider, temperature=temperature, **kwargs)
485- else:
486- response = await self._get_async_client().chat.completions.create(**kwargs)
487: return self._finish_summary(response)
488- except Exception as e:
489- delay = self._summary_attempt_failed(metrics, attempt, e)
490- if delay is None:
491: return _SUMMARY_FALLBACK
492- await asyncio.sleep(delay)
493-
494: def _plan_compression(self, trajectory: List[Dict[str, str]], metrics: TrajectoryMetrics) -> Optional[Tuple[int, int]]:
495- """Choose the ``[start, until)`` region to summarize, or None if nothing can be.
496-
497: Fills the pre-compression metrics either way. Accumulates turns from the
498: start of the compressible middle until the savings cover the overage plus
499- the summary itself, then snaps both boundaries off ``tool`` turns.
500- """
501- cfg = self.config
502- turn_tokens = self.count_turn_tokens(trajectory)
503- total_tokens = sum(turn_tokens)
504: metrics.original_turns = metrics.compressed_turns = len(trajectory)
505: metrics.original_tokens = metrics.compressed_tokens = total_tokens
506- if total_tokens <= cfg.target_max_tokens:
507- metrics.skipped_under_target = True
508- return None
509: metrics.still_over_limit = True
510- _, start, end = self._find_protected_indices(trajectory)
511- # Never *start* on an orphaned <tool_response> whose <tool_call> is in the protected head.
512- start = self._snap_boundary(trajectory, start, start, end)
513- if start >= end:
514- return None
515- # Replacing N turns with one summary saves sum(N) - summary_target_tokens.
516: target_tokens_to_compress = total_tokens - cfg.target_max_tokens + cfg.summary_target_tokens
517- accumulated = 0
518- until = start
519- for i in range(start, end):
520- accumulated += turn_tokens[i]
521- until = i + 1
522: if accumulated >= target_tokens_to_compress:
523- break
524: if accumulated < target_tokens_to_compress and until < end:
525- until = end
526- # The remainder is kept verbatim, so a tail boundary on a tool turn would orphan a marker.
527- until = self._snap_boundary(trajectory, until, start, end)
528- # A region no larger than the summary replacing it cannot shrink the trajectory.
529- if until <= start or sum(turn_tokens[start:until]) <= cfg.summary_target_tokens:
530- return None
531: metrics.turns_compressed_start_idx, metrics.turns_compressed_end_idx = start, until
532: metrics.turns_in_compressed_region = until - start
533- return start, until
534-
535: def _assemble_compressed(self, trajectory: List[Dict[str, str]], start: int, until: int, summary: str,
536- metrics: TrajectoryMetrics) -> List[Dict[str, str]]:
537- """Head (with summary notice on system) + summary human turn + verbatim tail; finalize metrics."""
538: compressed = []
539- for turn in trajectory[:start]:
540- turn = turn.copy()
541- if turn.get("from") == "system" and self.config.add_summary_notice:
542- turn["value"] = turn["value"] + self.config.summary_notice_text
543: compressed.append(turn)
544: compressed.append({"from": "human", "value": summary})
545: compressed.extend(turn.copy() for turn in trajectory[until:])
546: metrics.compressed_turns = len(compressed)
547: metrics.compressed_tokens = self.count_trajectory_tokens(compressed)
548: metrics.turns_removed = metrics.original_turns - metrics.compressed_turns
549: metrics.tokens_saved = metrics.original_tokens - metrics.compressed_tokens
550: metrics.compression_ratio = metrics.compressed_tokens / max(metrics.original_tokens, 1)
551: metrics.was_compressed = True
552: metrics.still_over_limit = metrics.compressed_tokens > self.config.target_max_tokens
553: return compressed
554-
555: def compress_trajectory(self, trajectory: List[Dict[str, str]]) -> Tuple[List[Dict[str, str]], TrajectoryMetrics]:
556: """Compress one trajectory into the target budget; returns ``(trajectory, metrics)``."""
557- metrics = TrajectoryMetrics()
558: region = self._plan_compression(trajectory, metrics)
559- if region is None:
560- return trajectory, metrics
561- summary = self._generate_summary(self._extract_turn_content_for_summary(trajectory, *region), metrics)
562: return self._assemble_compressed(trajectory, *region, summary, metrics), metrics
563-
564: async def compress_trajectory_async(self, trajectory: List[Dict[str, str]]) -> Tuple[List[Dict[str, str]], TrajectoryMetrics]:
565: """Async twin of ``compress_trajectory``."""
566- metrics = TrajectoryMetrics()
567: region = self._plan_compression(trajectory, metrics)
568- if region is None:
569- return trajectory, metrics
570- summary = await self._generate_summary_async(self._extract_turn_content_for_summary(trajectory, *region), metrics)
571: return self._assemble_compressed(trajectory, *region, summary, metrics), metrics
572-
573- async def process_entry_async(self, entry: Dict[str, Any]) -> Tuple[Dict[str, Any], TrajectoryMetrics]:
574: """Compress one JSONL entry's ``conversations``; attach metrics when compressed."""
575- if "conversations" not in entry:
576- return entry, TrajectoryMetrics()
577: compressed_trajectory, metrics = await self.compress_trajectory_async(entry["conversations"])
578: result = dict(entry, conversations=compressed_trajectory)
579: if self.config.metrics_per_trajectory and metrics.was_compressed:
580: result["compression_metrics"] = metrics.to_dict()
581- return result, metrics
582-
583- def process_directory(self, input_dir: Path, output_dir: Path):
584: """Compress every ``*.jsonl`` in ``input_dir`` into ``output_dir`` (async, parallel API calls)."""
585- asyncio.run(self._process_directory_async(input_dir, output_dir))
586-
587- async def _process_one(self, run: _RunProgress, file_path: Path, entry_idx: int, entry: Dict) -> Optional[Tuple[Dict[str, Any], TrajectoryMetrics]]:
588: """Process one entry under the semaphore/timeout; None means dropped (timed out)."""
589- async with run.semaphore:
590- async with run.lock:
591- run.in_flight += 1
592- try:
593: processed_entry, metrics = await asyncio.wait_for(self.process_entry_async(entry), timeout=self.config.per_trajectory_timeout)
594- async with run.lock:
595- self.aggregate_metrics.add_trajectory_metrics(metrics)
596: if metrics.was_compressed:
597: run.compressed += 1
598- run.api_calls += metrics.summarization_api_calls
599- run.skipped += bool(metrics.skipped_under_target)
600: run.finish()
601- return processed_entry, metrics
602: except asyncio.TimeoutError:
603: self.logger.warning("Timeout processing entry from %s:%s (>%ss)", file_path, entry_idx, self.config.per_trajectory_timeout)
604- async with run.lock:
605- self.aggregate_metrics.trajectories_failed += 1
606: run.timeouts += 1
607: run.finish()
608- return None
609- except Exception as e:
610- self.logger.error("Error processing entry from %s:%s: %s", file_path, entry_idx, e)
611- async with run.lock:
612- self.aggregate_metrics.trajectories_failed += 1
613: run.finish(update_status=False)
614- return entry, TrajectoryMetrics() # keep the original on error
615-
616- async def _process_directory_async(self, input_dir: Path, output_dir: Path):
617- console = Console()
618- self.aggregate_metrics.processing_start_time = datetime.now().isoformat()
619- start_time = time.time()
--
643- with Progress(
644- SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TaskProgressColumn(),
645- TextColumn("•"), TimeElapsedColumn(), TextColumn("•"), TimeRemainingColumn(),
646- console=console, refresh_per_second=10, # Higher refresh for async
647- ) as progress:
648- run = _RunProgress(
649: progress, progress.add_task(f"[cyan]Compressing {total_entries:,} trajectories", total=total_entries),
650- progress.add_task("[dim]Starting...[/dim]", total=None),
651- asyncio.Lock(), asyncio.Semaphore(self.config.max_concurrent_requests),
652- )
653- outcomes = await asyncio.gather(*(self._process_one(run, *item) for item in all_entries))
654- progress.remove_task(run.status_task)
655-
--
670- metrics_path = output_dir / self.config.metrics_output_file
671- with open(metrics_path, 'w', encoding="utf-8") as f:
672- json.dump(self.aggregate_metrics.to_dict(), f, indent=2)
673- console.print(f"\n💾 Metrics saved to {metrics_path}")
674-
675- def _print_summary(self):
676: """Print comprehensive compression summary statistics."""
677- m = self.aggregate_metrics.to_dict()
678- s, t, u, a, z, p = m['summary'], m['tokens'], m['turns'], m['averages'], m['summarization'], m['processing']
679: total, compressed = s['total_trajectories'], s['trajectories_compressed']
680- pct = lambda n: (n / max(total, 1)) * 100 # noqa: E731
681- duration = p['duration_seconds']
682- time_str = f"{duration/60:.1f} minutes" if duration > 60 else f"{duration:.1f} seconds"
683-
684- sections = [
685- ("📁 TRAJECTORIES", 54, [
686- f"║{'':4}Total Processed: {total:>10,}{' '*32}║",
687: f"║{'':4}├─ Compressed: {compressed:>10,} ({pct(compressed):>5.1f}%){' '*18}║",
688: f"║{'':4}├─ Skipped (under limit):{s['trajectories_skipped_under_target']:>9,} ({pct(s['trajectories_skipped_under_target']):>5.1f}%){' '*18}║",
689: f"║{'':4}├─ Still over limit: {s['trajectories_still_over_limit']:>10,} ({pct(s['trajectories_still_over_limit']):>5.1f}%){' '*18}║",
690- f"║{'':4}└─ Failed: {s['trajectories_failed']:>10,}{' '*32}║",
691- ]),
692- ("🔢 TOKENS", 60, [
693: f"║{'':4}Before Compression: {t['total_before']:>15,} tokens{' '*21}║",
694: f"║{'':4}After Compression: {t['total_after']:>15,} tokens{' '*21}║",
695- f"║{'':4}Total Saved: {t['total_saved']:>15,} tokens{' '*21}║",
696: f"║{'':4}Overall Compression: {t['overall_compression_ratio']:>14.1%}{' '*28}║",
697- ] + ([f"║{'':4}Space Savings: {(t['total_saved'] / t['total_before']) * 100:>14.1f}%{' '*28}║"] if t['total_before'] > 0 else [])),
698- ("💬 CONVERSATION TURNS", 48, [
699: f"║{'':4}Before Compression: {u['total_before']:>15,} turns{' '*22}║",
700: f"║{'':4}After Compression: {u['total_after']:>15,} turns{' '*22}║",
701- f"║{'':4}Total Removed: {u['total_removed']:>15,} turns{' '*22}║",
702- ]),
703: ("📈 AVERAGES (Compressed Trajectories Only)", 27, [
704: f"║{'':4}Avg Compression Ratio: {a['avg_compression_ratio']:>14.1%}{' '*28}║",
705: f"║{'':4}Avg Tokens Saved: {a['avg_tokens_saved_per_compressed']:>14,.0f}{' '*28}║",
706: f"║{'':4}Avg Turns Removed: {a['avg_turns_removed_per_compressed']:>14.1f}{' '*28}║",
707: ] if compressed > 0 else [f"║{'':4}No trajectories were compressed{' '*38}║"]),
708- ("🤖 SUMMARIZATION API", 49, [
709- f"║{'':4}API Calls Made: {z['total_api_calls']:>15,}{' '*27}║",
710- f"║{'':4}Errors: {z['total_errors']:>15,}{' '*27}║",
711- f"║{'':4}Success Rate: {z['success_rate']:>14.1%}{' '*28}║",
712- ]),
713- ("⏱️ PROCESSING TIME", 51, [
714- f"║{'':4}Duration: {time_str:>20}{' '*22}║",
715- f"║{'':4}Throughput: {total / max(duration, 0.001):>15.1f} traj/sec{' '*18}║",
716- f"║{'':4}Started: {p['start_time'][:19]:>20}{' '*22}║",
717: f"║{'':4}Finished: {p['end_time'][:19]:>20}{' '*22}║",
718- ]),
719- ]
720- print("\n")
721- print(f"╔{'═'*70}╗")
722: print(f"║{'TRAJECTORY COMPRESSION REPORT':^70}║")
723- for title, pad, rows in sections:
724- print(f"╠{'═'*70}╣")
725- print(f"║{'':2}{title}{' '*pad}║")
726- print(f"║{'─'*70}║")
727- for row in rows:
728- print(row)
729- print(f"╚{'═'*70}╝")
730-
731: ratios = self.aggregate_metrics.compression_ratios
732- if ratios:
733- saved = self.aggregate_metrics.tokens_saved_list
734- print("\n📊 Distribution Summary:")
735: print(f" Compression ratios: min={min(ratios):.2%}, max={max(ratios):.2%}, median={sorted(ratios)[len(ratios)//2]:.2%}")
736- print(f" Tokens saved: min={min(saved):,}, max={max(saved):,}, median={sorted(saved)[len(saved)//2]:,}")
737-
738-
739-# ---------------------------------------------------------------------------
740-# CLI
741-# ---------------------------------------------------------------------------
742-
743:def _load_cli_config(config: str, target_max_tokens: Optional[int], tokenizer: Optional[str]) -> CompressionConfig:
744- """Load the YAML config (defaults if missing) and apply CLI overrides."""
745- if Path(config).exists():
746- print(f"📋 Loading config from {config}")
747: compression_config = CompressionConfig.from_yaml(config)
748- else:
749- print(f"⚠️ Config not found at {config}, using defaults")
750: compression_config = CompressionConfig()
751- if target_max_tokens:
752: compression_config.target_max_tokens = target_max_tokens
753- if tokenizer:
754: compression_config.tokenizer_name = tokenizer
755: return compression_config
756-
757-
758-def _print_dry_run(icon: str, target: Any, output_path: Path) -> None:
759- print("\n🔍 DRY RUN MODE - analyzing without writing")
760- print(f"{icon} Would process: {target}")
761- print(f"{icon} Would output to: {output_path}")
762-
763-
764-def _sample(entries: list, sample_percent: float) -> list:
765- return random.sample(entries, min(max(1, int(len(entries) * sample_percent / 100)), len(entries)))
766-
767-
768:def _run_file_mode(input_path: Path, output: Optional[str], compression_config: CompressionConfig, sample_percent: Optional[float], seed: int, dry_run: bool) -> None:
769: """Single-file input: (sample,) compress via a temp directory, merge into one output file."""
770- print("📄 Input mode: Single JSONL file")
771: output_path = Path(output) if output else input_path.parent / (input_path.stem + compression_config.output_suffix + ".jsonl")
772- entries = [entry for _, entry in _load_jsonl(input_path, lambda n, e: print(f"⚠️ Skipping invalid JSON at line {n}: {e}"), start=1)]
773- total_entries = len(entries)
774- print(f" Loaded {total_entries:,} trajectories from {input_path.name}")
775- if sample_percent is not None:
776- random.seed(seed)
777- entries = random.sample(entries, max(1, int(total_entries * sample_percent / 100)))
--
781- return
782-
783- with tempfile.TemporaryDirectory() as temp_dir:
784- temp_input_dir, temp_output_dir = Path(temp_dir) / "input", Path(temp_dir) / "output"
785- temp_input_dir.mkdir()
786- _write_jsonl(temp_input_dir / "trajectories.jsonl", entries)
787: TrajectoryCompressor(compression_config).process_directory(temp_input_dir, temp_output_dir)
788- output_path.parent.mkdir(parents=True, exist_ok=True)
789- with open(output_path, 'w', encoding='utf-8') as out_f:
790- for jsonl_file in sorted(temp_output_dir.glob("*.jsonl")):
791- with open(jsonl_file, 'r', encoding='utf-8') as in_f:
792- shutil.copyfileobj(in_f, out_f)
793: metrics_file = temp_output_dir / compression_config.metrics_output_file
794- if metrics_file.exists():
795- metrics_output = output_path.parent / (output_path.stem + "_metrics.json")
796- shutil.copy(metrics_file, metrics_output)
797- print(f"💾 Metrics saved to {metrics_output}")
798: print("\n✅ Compression complete!")
799- print(f"📄 Output: {output_path}")
800-
801-
802:def _run_dir_mode(input_path: Path, output: Optional[str], compression_config: CompressionConfig, sample_percent: Optional[float], seed: int, dry_run: bool) -> None:
803: """Directory input: compress in place, or per-file sample into a temp dir first."""
804- print("📁 Input mode: Directory of JSONL files")
805: output_path = Path(output) if output else input_path.parent / (input_path.name + compression_config.output_suffix)
806- if sample_percent is None:
807- if dry_run:
808- _print_dry_run("📁", input_path, output_path)
809- return
810: TrajectoryCompressor(compression_config).process_directory(input_path, output_path)
811- else:
812- print(f"\n⚠️ Sampling from directory: will sample {sample_percent}% from each file")
813- with tempfile.TemporaryDirectory() as temp_dir:
814- temp_input_dir = Path(temp_dir) / "input"
815- temp_input_dir.mkdir()
816- random.seed(seed)
--
822- total_sampled += len(sampled_entries)
823- _write_jsonl(temp_input_dir / jsonl_file.name, sampled_entries)
824- print(f" Sampled {total_sampled:,} from {total_original:,} total trajectories")
825- if dry_run:
826- _print_dry_run("📁", temp_input_dir, output_path)
827- return
828: TrajectoryCompressor(compression_config).process_directory(temp_input_dir, output_path)
829: print("\n✅ Compression complete!")
830-
831-
832:def main(input: str, output: str = None, config: str = "configs/trajectory_compression.yaml", target_max_tokens: int = None,
833- tokenizer: str = None, sample_percent: float = None, seed: int = 42, dry_run: bool = False):
834- """
835: Compress agent trajectories to fit within a target token budget.
836-
837- Supports both single JSONL files and directories containing multiple JSONL files.
838: Optionally sample a percentage of trajectories before compression.
839-
840- Args:
841- input: Path to JSONL file or directory containing JSONL files
842- output: Output path (file for file input, directory for dir input)
843: Default: adds "_compressed" suffix to input name
844- config: Path to YAML configuration file
845- target_max_tokens: Override target token count from config
846- tokenizer: Override tokenizer name from config
847: sample_percent: Sample this percentage of trajectories (1-100) before compression
848- seed: Random seed for sampling reproducibility (default: 42)
849: dry_run: Analyze without compressing (just show what would happen)
850- """
851: print("🗜️ Trajectory Compressor")
852- print("=" * 60)
853: compression_config = _load_cli_config(config, target_max_tokens, tokenizer)
854- if sample_percent is not None:
855- if sample_percent <= 0 or sample_percent > 100:
856- print(f"❌ sample_percent must be between 1 and 100, got {sample_percent}")
857- return
858- print(f"🎲 Will sample {sample_percent}% of trajectories (seed={seed})")
859- input_path = Path(input)
860- if not input_path.exists():
861- print(f"❌ Input not found: {input}")
862- return
863- run_mode = _run_file_mode if input_path.is_file() else _run_dir_mode
864: run_mode(input_path, output, compression_config, sample_percent, seed, dry_run)
865-
866-
867-if __name__ == "__main__":
868- fire.Fire(main)
===== /home/hermes/.hermes/hermes-agent/utils.py =====
5-import logging
6-import os
7-import shutil
8-import stat
9-import tempfile
10-import time
11:from contextlib import suppress
12-from pathlib import Path
13-from typing import Any, Union
14-from urllib.parse import urlparse
15-
16-import yaml
17-
--
81-# without FILE_SHARE_DELETE, so ``os.replace`` onto an open file is denied with 5 ERROR_ACCESS_D