コンテキスト プロバイダーは、各呼び出しを実行して実行前にコンテキストを追加し、実行後にデータを処理します。
組み込みパターン
通常のパターンでは、エージェントの作成時に 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])
Important
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])