Поставщики контекстов

Поставщики контекста запускаются при каждом вызове для добавления контекста перед исполнением и обработки данных после исполнения.

Встроенный шаблон

Обычно провайдеры настраиваются через context_providers=[...] при создании агента.

ChatHistoryProvider и AIContextProvider являются встроенными точками расширения для краткосрочной истории и долгосрочного обогащения контекста.

Для Python InMemoryHistoryProvider используется встроенный поставщик истории, применяемый для локальной памяти общения.

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 может автоматически добавлять InMemoryHistoryProvider("memory") в некоторых случаях, но добавьте его явно, если требуется предсказуемое поведение локальной памяти.

Поставщик пользовательского контекста

Используйте настраиваемые поставщики контекстов, если необходимо внедрить динамические инструкции или сообщения или извлечь состояние после выполнения.

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

Провайдер пользовательской истории действий

Поставщики истории — это поставщики контекста, специализированные для загрузки и хранения сообщений.

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

Это важно

В Python можно настроить несколько поставщиков истории, но только один из них должен использовать load_messages=True. Используйте дополнительных поставщиков для диагностики и оценки с load_messages=False и store_context_messages=True, чтобы они могли фиксировать контекст от других поставщиков вместе с входными и выходными данными.

Пример шаблона:

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

Дальнейшие шаги

Хранение