Kontextanbieter

Kontextanbieter werden um jeden Aufruf herum ausgeführt, um den Kontext vor der Ausführung hinzuzufügen und die Daten nach der Ausführung zu verarbeiten.

Integriertes Muster

Die übliche Vorgehensweise besteht darin, Anbieter über context_providers=[...] beim Erstellen eines Agents zu konfigurieren.

ChatHistoryProvider und AIContextProvider sind die integrierten Erweiterungspunkte für die kurzfristige Historie und langfristige kontextuelle Anreicherung.

Für Python ist der integrierte Verlaufsanbieter InMemoryHistoryProvider für den lokalen Konversationsspeicher zuständig.

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 kann InMemoryHistoryProvider("memory") in bestimmten Fällen automatisch hinzufügen, aber fügen Sie es jedoch explizit hinzu, wenn Sie ein deterministisches lokales Speicherverhalten wünschen.

Benutzerdefinierter Kontextanbieter

Verwenden Sie benutzerdefinierte Kontextanbieter, wenn Sie dynamische Anweisungen/Nachrichten einfügen oder den Zustand nach ausführung extrahieren müssen.

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

Benutzerdefinierter Verlaufsanbieter

Verlaufsanbieter sind Kontextanbieter, die auf das Laden/Speichern von Nachrichten spezialisiert sind.

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])

Von Bedeutung

In Python können Sie mehrere Verlaufsanbieter konfigurieren, aber nur einer sollte verwendet werden load_messages=True. Verwenden Sie zusätzliche Anbieter für Diagnosen/Evaluierungen mit load_messages=False und store_context_messages=True, damit sie den Kontext von anderen Anbietern zusammen mit Eingaben/Ausgaben erfassen.

Beispielmuster:

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

Nächste Schritte

Lagerung