Proveedores de contexto

Los proveedores de contexto se ejecutan en torno a cada invocación para agregar contexto antes de la ejecución y procesar los datos después de la ejecución.

Patrón incorporado

El patrón habitual es configurar proveedores a través de context_providers=[...] al crear un agente.

ChatHistoryProvider y AIContextProvider son los puntos de extensión integrados para el historial a corto plazo y el enriquecimiento a largo plazo/contexto.

Para Python, InMemoryHistoryProvider es el proveedor de historial integrado que se usa para la memoria conversacional local.

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 puede agregar InMemoryHistoryProvider("memory") automáticamente en casos específicos, pero házlo explícitamente cuando desees un comportamiento de memoria local determinista.

Proveedor de contexto personalizado

Use proveedores de contexto personalizados cuando necesite insertar instrucciones o mensajes dinámicos o extraer el estado después de las ejecuciones.

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(".")

Proveedor de historial personalizado

Los proveedores de historial son proveedores de contexto especializados para cargar o almacenar mensajes.

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

En Python, puede configurar varios proveedores de historial, pero solo uno debe usar load_messages=True. Use proveedores adicionales para diagnósticos o evaluaciones con load_messages=False y store_context_messages=True para que capturen el contexto de otros proveedores junto con la entrada y salida.

Patrón de ejemplo:

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

Pasos siguientes

Almacenamiento