Method ingest

python/valuecell/core/event/buffer.py:145–216

Ingest a response and return a list of SaveItem objects to persist. Depending on the event type this will either: - Flush and emit an immediate item (for immediate events), or - Accumulate buffered chunks and emit an upsert SaveItem with the current aggregated payl

(self, resp: BaseResponse)

        return resp

    def ingest(self, resp: BaseResponse) -> List[SaveItem]:
        """Ingest a response and return a list of SaveItem objects to persist.

        Depending on the event type this will either:
        - Flush and emit an immediate item (for immediate events), or
        - Accumulate buffered chunks and emit an upsert SaveItem with the
          current aggregated payload for the paragraph entry.

        Returns:
            A list of SaveItem objects that should be persisted by the caller.
        """
        data: UnifiedResponseData = resp.data
        ev = resp.event

        ctx = (
            data.conversation_id,
            data.thread_id,
            data.task_id,
        )
        out: List[SaveItem] = []

        # Immediate: write-through, but treat as paragraph boundary for buffered keys
        if ev in self._immediate_events:
            # Flush buffered aggregates for this context before the immediate item
            conv_id, th_id, tk_id = ctx
            keys_to_flush = self._collect_task_keys(conv_id, th_id, tk_id)
            out.extend(self._finalize_keys(keys_to_flush))
            # Now write the immediate item
            out.append(self._make_save_item_from_response(resp))
            return out

        # Buffered: accumulate by (ctx + event)
        if ev in self._buffered_events:
            key: BufferKey = (*ctx, ev)
            entry = self._buffers.get(key)
            if not entry:
                # If annotate() wasn't called, create an entry now.
                entry = BufferEntry(role=data.role, agent_name=data.agent_name)
                self._buffers[key] = entry
            elif entry.agent_name is None and data.agent_name:
                entry.agent_name = data.agent_name

            # Extract text content from payload
            payload = data.payload
            text = None
            if isinstance(payload, BaseResponseDataPayload):
                text = payload.content or ""
            elif isinstance(payload, BaseModel):
                # Fallback: serialize whole payload
                text = payload.model_dump_json(exclude_none=True)
            elif isinstance(payload, str):
                text = payload
            else:
                text = ""

            if text:
                entry.append(text)
                # Always upsert current aggregate (no size-based rotation)