Bağlam Sağlayıcıları

Bağlam sağlayıcıları yürütmeden önce bağlam eklemek ve yürütmeden sonra verileri işlemek için her çağrının etrafında çalışır.

Yerleşik desen

Normal düzen, bir aracı oluştururken context_providers=[...] üzerinden sağlayıcıları yapılandırmaktır.

ChatHistoryProvider ve AIContextProvider kısa vadeli geçmiş ve uzun vadeli/bağlam zenginleştirmesi için yerleşik uzantı noktalarıdır.

Python için, InMemoryHistoryProvider yerel konuşma belleği için kullanılan yerleşik geçmiş sağlayıcısıdır.

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 belirli durumlarda otomatik olarak ekleyebilir InMemoryHistoryProvider("memory") , ancak belirleyici yerel bellek davranışı istediğinizde açıkça ekleyebilirsiniz.

Özel bağlam sağlayıcısı

Çalıştırmalardan sonra dinamik yönergeler/iletiler eklemeniz veya durum ayıklamanız gerektiğinde özel bağlam sağlayıcılarını kullanın.

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

Özel geçmiş sağlayıcısı

Geçmiş sağlayıcıları, iletileri yüklemek/depolamak için özelleştirilmiş bağlam sağlayıcılarıdır.

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

Önemli

Python'da birden çok geçmiş sağlayıcısı yapılandırabilirsiniz, ancak yalnızca birinin kullanması load_messages=Truegerekir. Tanılama/değerlendirmeler için load_messages=False ve store_context_messages=True ile ek sağlayıcılar kullanın; böylece giriş/çıkış ile birlikte diğer sağlayıcılardan da bağlamı yakalarlar.

Örnek desen:

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

Sonraki Adımlar

Depolama