上下文提供程序围绕每个调用运行，在执行前添加上下文，并在执行后处理数据。
内置模式
常规模式是在创建代理时，通过 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])