Context Providers

Contextproviders worden rond elke aanroep uitgevoerd om context toe te voegen vóór uitvoering en gegevens te verwerken na uitvoering.

Ingebouwd patroon

Het normale patroon is het configureren van providers bij het maken van een agent via context_providers=[...].

ChatHistoryProvider en AIContextProvider zijn de ingebouwde uitbreidingspunten voor korte termijngeschiedenis en langetermijn-/contextverrijking.

Voor Python InMemoryHistoryProvider is de ingebouwde geschiedenisprovider die wordt gebruikt voor lokaal gespreksgeheugen.

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 kan automatisch toevoegen InMemoryHistoryProvider("memory") in specifieke gevallen, maar voeg deze expliciet toe wanneer u het gedrag van het deterministische lokale geheugen wilt.

Aangepaste context-provider

Gebruik aangepaste contextproviders wanneer u dynamische instructies/berichten moet invoeren of de status na uitvoeringen moet extraheren.

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

Aangepaste geschiedenisprovider

Geschiedenisproviders zijn contextproviders die gespecialiseerd zijn voor het laden/opslaan van berichten.

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

Belangrijk

In Python kunt u meerdere geschiedenisproviders configureren, maar slechts één zou load_messages=True moeten gebruiken. Gebruik extra providers voor diagnostische evaluaties met load_messages=False en store_context_messages=True, zodat ze naast invoer/uitvoer ook de context van andere providers vastleggen.

Voorbeeldpatroon:

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

Volgende stappen

Storage