上下文工程

class ContextEngine:
    """
    Manages the lifecycle and processing of conversational context.

    ContextEngine acts as the central entry-point for:
    1. Registering and configuring message processors.
    2. Creating isolated ModelContext instances tied to a session.
    3. Applying processor chains to enforce window limits, compression, etc.

    Parameters
    ----------
    config : ContextEngineConfig, optional
        Global engine settings (message/token limits, processor defaults).
        If omitted, a default configuration is used.
    """

async def compress_context(
        self,
        context_id: str = "default_context_id",
        session: Session = None,
        *,
        session_id: str = None,
        processor_types: List[str] = None,
        **kwargs,
) -> str | dict[str, Any]:
    """
    Actively run registered compression processors for an existing context.

    Returns:
        Compression result code:
        - ``"busy"``: passive compression is already in progress.
        - ``"compressed"``: active compression ran and changed context.
        - ``"noop"``: active compression ran but nothing changed, or no
          compression processor is registered.

    A successful compression is saved to the resolved session and committed
    before this method returns. Contexts without a session remain in-memory only.
    """

@_fw.emit_after(ContextEvents.CONTEXT_OFFLOADED, result_key="result")
async def save_contexts(self,
                        session: Session,
                        context_ids: List[str] = None
                        ):
    """
    Batch-persist multiple contexts and their runtime states.

    Each context's messages, sliding-window position, token count and statistics
    are saved locally.
    """
    if not session:
        context_engine_logger.warning(
            "Save context failed, session cannot be None",
            event_type=LogEventType.CONTEXT_SAVE,
        )
        return
    session_id = session.get_session_id()
    states = dict()
    if context_ids is None:
        context_ids = [
            context.context_id() for context_id, context in self._context_pool.items()
            if context.session_id() == session_id
        ]

    for context_id in context_ids:
        context_id = self._process_context_id(context_id)
        full_context_id = f"{session_id}_{context_id}"
        context = self._context_pool.get(full_context_id)
        if context is None or not hasattr(context, "save_state"):
            continue
        context_state = context.save_state()
        states[context_id] = context_state
    self._save_state_to_session(session, states)
    return states