Function _summarize_events_with_trace

src/google/adk/apps/compaction.py:35–63

Summarizes events within a trace span labeled for compaction.

(
    *,
    session: Session,
    config: EventsCompactionConfig,
    events_to_compact: list[Event],
    trigger: str,
)

async def _summarize_events_with_trace(
    *,
    session: Session,
    config: EventsCompactionConfig,
    events_to_compact: list[Event],
    trigger: str,
) -> Event | None:
  """Summarizes events within a trace span labeled for compaction."""
  if config.summarizer is None:
    return None

  attributes = _build_compaction_attributes(
      session_id=session.id,
      trigger=trigger,
      summarizer_type=type(config.summarizer).__name__,
      event_count=len(events_to_compact),
      token_threshold=config.token_threshold,
      event_retention_size=config.event_retention_size,
      compaction_interval=config.compaction_interval,
      overlap_size=config.overlap_size,
  )

  with tracer.start_as_current_span(f'compact_events {trigger}') as span:
    span.set_attributes(attributes)
    compaction_event = await config.summarizer.maybe_summarize_events(
        events=events_to_compact
    )
    span.set_attributes(_build_compaction_result_attributes(compaction_event))
    return compaction_event


def _count_text_chars_in_content(content: types.Content | None) -> int: