Penyedia Konteks

Penyedia konteks menjalankan setiap pemanggilan untuk menambahkan konteks sebelum eksekusi dan memproses data setelah eksekusi.

Pola bawaan

Pola umum adalah mengonfigurasi penyedia melalui context_providers=[...] saat membuat agen.

ChatHistoryProvider dan AIContextProvider merupakan titik ekstensi bawaan untuk riwayat jangka pendek dan pengayaan jangka panjang/konteks.

Untuk Python, InMemoryHistoryProvider adalah penyedia riwayat bawaan yang digunakan untuk memori percakapan lokal.

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 dapat menambahkan InMemoryHistoryProvider("memory") secara otomatis dalam kasus tertentu, tetapi menambahkannya secara eksplisit saat Anda menginginkan perilaku memori lokal deterministik.

Penyedia konteks kustom

Gunakan penyedia konteks kustom saat Anda perlu menyuntikkan instruksi/pesan dinamis atau mengekstrak status setelah dijalankan.

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

Penyedia riwayat kustom

Penyedia riwayat adalah penyedia konteks khusus untuk memuat/menyimpan pesan.

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

Penting

Di Python, Anda dapat mengonfigurasi beberapa penyedia riwayat, tetapi hanya satu yang harus menggunakan load_messages=True. Gunakan penyedia tambahan untuk diagnostik/evaluasi dengan load_messages=False dan store_context_messages=True sehingga mereka mengumpulkan konteks dari penyedia lain bersama dengan input/output.

Contoh pola:

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

Langkah selanjutnya

Penyimpanan