Skip to content

symfonic.core.learning.phases_episodic

phases_episodic

Episodic consolidation phase implementations.

Phase 10 lives here because it operates on the EpisodicLayer (vector-backed) rather than the GraphMemoryStore used by phases 1-9 in phases.py.

summarize_episodic async 

summarize_episodic(
    episodic_layer: Any,
    scope: Any,
    *,
    llm_summarise: Any | None = None,
    max_entries: int = 100,
    summarize_batch: int = 50,
    gate: DurabilityGate | None = None,
) -> tuple[str, int]

Summarize oldest episodic entries into a single text, delete originals.

When the total event count exceeds max_entries, the oldest summarize_batch entries are retrieved, concatenated (or LLM- summarised when llm_summarise is provided), and the originals are deleted from the vector store.

Parameters:

Name Type Description Default
episodic_layer Any

An EpisodicLayer instance.

 required
scope Any

Tenant scope for isolation.

 required
llm_summarise Any | None

Optional async callable (text) -> summary.

 None
max_entries int

Threshold below which summarization is skipped.

 100
summarize_batch int

Number of oldest entries to summarize per run.

 50

Returns:

Type Description
str

Tuple of (summary_text, count_deleted). When no
int

summarization was needed, returns ("", 0).
Source code in src/symfonic/core/learning/phases_episodic.py 
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
async def summarize_episodic(
    episodic_layer: Any,
    scope: Any,
    *,
    llm_summarise: Any | None = None,
    max_entries: int = 100,
    summarize_batch: int = 50,
    gate: DurabilityGate | None = None,
) -> tuple[str, int]:
    """Summarize oldest episodic entries into a single text, delete originals.

    When the total event count exceeds ``max_entries``, the oldest
    ``summarize_batch`` entries are retrieved, concatenated (or LLM-
    summarised when ``llm_summarise`` is provided), and the originals
    are deleted from the vector store.

    Args:
        episodic_layer: An ``EpisodicLayer`` instance.
        scope: Tenant scope for isolation.
        llm_summarise: Optional async callable ``(text) -> summary``.
        max_entries: Threshold below which summarization is skipped.
        summarize_batch: Number of oldest entries to summarize per run.

    Returns:
        Tuple of ``(summary_text, count_deleted)``.  When no
        summarization was needed, returns ``("", 0)``.
    """
    count = await episodic_layer.count_events(scope)
    if count <= max_entries:
        return ("", 0)

    # Retrieve oldest entries.  list_events() returns results in
    # cosine-descending order (arbitrary for zero-vectors), NOT time order.
    # Phase 10 requires the *oldest* ``summarize_batch`` entries, so we
    # fetch all entries (up to ``count``), sort by created_at ascending,
    # and slice the front.  created_at is stored in metadata by store_event().
    all_entries = await episodic_layer.list_events(scope, limit=count)
    all_entries = sorted(all_entries, key=lambda e: e.metadata.get("created_at", ""))
    entries = all_entries[:summarize_batch]
    if not entries:
        return ("", 0)

    # Build raw text from entry contents.  v7.26.2: transient/expired
    # entries are gated out of the summary so they never become a
    # semantic meta-node.  They are NOT deleted here -- Phase 8 (cleanup)
    # owns their pruning; Phase 10 only refuses to promote them.
    _gate = gate if gate is not None else DurabilityGate()
    raw_parts: list[str] = []
    summarised_entries = []
    for entry in entries:
        if _gate.is_gated(entry, phase="10"):
            continue
        raw_parts.append(entry.content or "")
        summarised_entries.append(entry)
    raw_text = "\n".join(part for part in raw_parts if part)

    # Summarize
    if llm_summarise is not None:
        try:
            summary_text: str = await llm_summarise(raw_text)
        except Exception:
            logger.warning("LLM episodic summarization failed, using raw concat")
            summary_text = raw_text
    else:
        summary_text = raw_text

    # Delete originals from vector store -- only the entries actually
    # folded into the summary.  Gated (transient/expired) entries are left
    # in place for Phase 8 to prune.
    entry_ids = [entry.id for entry in summarised_entries if entry.id]
    if entry_ids:
        await episodic_layer._vector.delete(scope, entry_ids)

    deleted = len(entry_ids)
    logger.info(
        "Episodic summarization: summarized %d entries for tenant=%s",
        deleted,
        scope.tenant_id,
    )

    return (summary_text, deleted)