Provider di contesto

I provider di contesto vengono eseguiti in occasione di ogni invocazione per aggiungere contesto prima dell'esecuzione ed elaborare i dati dopo l'esecuzione.

Modello predefinito

Il modello regolare consiste nel configurare i provider tramite context_providers=[...] durante la creazione di un agente.

ChatHistoryProvider e AIContextProvider sono i punti di estensione predefiniti per la cronologia a breve termine e l'arricchimento a lungo termine/contesto.

Per Python, InMemoryHistoryProvider è il provider di cronologia predefinito usato per la memoria conversazionale locale.

from agent_framework import InMemoryHistoryProvider
from agent_framework.openai import OpenAIChatClient

agent = OpenAIChatClient().as_agent(
    name="MemoryBot",
    instructions="You are a helpful assistant.",
    context_providers=[InMemoryHistoryProvider("memory", load_messages=True)],
)

session = agent.create_session()
await agent.run("Remember that I prefer vegetarian food.", session=session)

RawAgent può aggiungere InMemoryHistoryProvider("memory") automaticamente in casi specifici, ma aggiungerlo in modo esplicito quando si desidera un comportamento di memoria locale deterministico.

Provider di contesto personalizzato

Usare provider di contesto personalizzati quando è necessario inserire istruzioni/messaggi dinamici o estrarre lo stato dopo l'esecuzione.

from typing import Any

from agent_framework import AgentSession, BaseContextProvider, SessionContext


class UserPreferenceProvider(BaseContextProvider):
    def __init__(self) -> None:
        super().__init__("user-preferences")

    async def before_run(
        self,
        *,
        agent: Any,
        session: AgentSession,
        context: SessionContext,
        state: dict[str, Any],
    ) -> None:
        if favorite := state.get("favorite_food"):
            context.extend_instructions(self.source_id, f"User's favorite food is {favorite}.")

    async def after_run(
        self,
        *,
        agent: Any,
        session: AgentSession,
        context: SessionContext,
        state: dict[str, Any],
    ) -> None:
        for message in context.input_messages:
            text = (message.text or "") if hasattr(message, "text") else ""
            if isinstance(text, str) and "favorite food is" in text.lower():
                state["favorite_food"] = text.split("favorite food is", 1)[1].strip().rstrip(".")

Provider di cronologia personalizzato

I provider di cronologia sono provider di contesto specializzati per il caricamento o l'archiviazione dei messaggi.

from collections.abc import Sequence
from typing import Any

from agent_framework import BaseHistoryProvider, Message


class DatabaseHistoryProvider(BaseHistoryProvider):
    def __init__(self, db: Any) -> None:
        super().__init__("db-history", load_messages=True)
        self._db = db

    async def get_messages(
        self,
        session_id: str | None,
        *,
        state: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> list[Message]:
        key = (state or {}).get(self.source_id, {}).get("history_key", session_id or "default")
        rows = await self._db.load_messages(key)
        return [Message.from_dict(row) for row in rows]

    async def save_messages(
        self,
        session_id: str | None,
        messages: Sequence[Message],
        *,
        state: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:
        if not messages:
            return
        if state is not None:
            key = state.setdefault(self.source_id, {}).setdefault("history_key", session_id or "default")
        else:
            key = session_id or "default"
        await self._db.save_messages(key, [m.to_dict() for m in messages])

Importante

In Python è possibile configurare più provider di cronologia, ma solo uno deve usare load_messages=True. Usare provider aggiuntivi per la diagnostica/valutazione con load_messages=False e store_context_messages=True in modo da acquisire il contesto da altri provider insieme all'input/output.

Modello di esempio:

primary = DatabaseHistoryProvider(db)
audit = InMemoryHistoryProvider("audit", load_messages=False, store_context_messages=True)
agent = OpenAIChatClient().as_agent(context_providers=[primary, audit])

Passaggi successivi

Spazio di archiviazione