Zprostředkovatelé kontextu

Zprostředkovatelé kontextu se spouští kolem každého vyvolání, aby před spuštěním a zpracování dat po spuštění přidali kontext.

Poznámka:

Seznam předem připravených zprostředkovatelů kontextu, které můžete použít se svým agentem, najdete v tématu Integrace.

Předdefinovaný vzor

Při vytváření agenta nakonfigurujte zprostředkovatele prostřednictvím možností konstruktoru. AIContextProvider je integrovaný bod rozšíření pro rozšiřování paměti a kontextu.

AIAgent agent = new OpenAIClient("<your_api_key>")
    .GetChatClient(modelName)
    .AsAIAgent(new ChatClientAgentOptions()
    {
        ChatOptions = new() { Instructions = "You are a helpful assistant." },
        AIContextProviders = [
            new MyCustomMemoryProvider()
        ],
    });

AgentSession session = await agent.CreateSessionAsync();
Console.WriteLine(await agent.RunAsync("Remember my name is Alice.", session));

Návod

Seznam předdefinovaných AIContextProvider implementací najdete v tématu Integrace.

Běžným postupem je konfigurovat poskytovatele prostřednictvím context_providers=[...] při vytváření agenta.

InMemoryHistoryProvider je integrovaný zprostředkovatel historie používaný pro místní konverzační paměť.

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 může automaticky přidávat InMemoryHistoryProvider("memory") v konkrétních případech, ale přidat ho explicitně, pokud chcete deterministické chování místní paměti.

Vlastní zprostředkovatel kontextu

Vlastní zprostředkovatele kontextu použijte v případě, že po spuštění potřebujete vložit dynamické instrukce, zprávy nebo nástroje nebo extrahovat stav.

Základní třída pro zprostředkovatele kontextu je Microsoft.Agents.AI.AIContextProvider. Zprostředkovatelé kontextu se účastní kanálu agenta, mají možnost přispívat do vstupních zpráv agenta nebo je přepsat a mohou extrahovat informace z nových zpráv. AIContextProvider má různé virtuální metody, které můžete přepsat pro implementaci vlastního zprostředkovatele kontextu. Další informace o tom, co je potřeba přepsat, jsou uvedeny v možnostech implementace níže.

AIContextProvider Stav

instance AIContextProvider je připojena k agentovi a stejná instance bude použita pro všechny relace. To znamená, že AIContextProvider v instanci poskytovatele by se neměl ukládat žádný konkrétní stav relace. Může AIContextProvider mít odkaz na klienta služby paměti v poli, ale neměl by mít ID pro konkrétní sadu pamětí v poli.

AIContextProvider může místo toho ukládat jakékoli hodnoty specifické pro relaci, jako jsou ID paměti, zprávy nebo cokoli jiného, co je relevantní pro AgentSession. Všem virtuálním metodám AIContextProvider se předává odkaz na aktuální AIAgent a AgentSession.

Chcete-li snadno ukládat stav s určeným typem v AgentSession, je k dispozici pomocná třída:

// First define a type containing the properties to store in state
internal class MyCustomState
{
    public string? MemoryId { get; set; }
}

// Create the helper
var sessionStateHelper = new ProviderSessionState<MyCustomState>(
    // stateInitializer is called when there is no state in the session for this AIContextProvider yet
    stateInitializer: currentSession => new MyCustomState() { MemoryId = Guid.NewGuid().ToString() },
    // The key under which to store state in the session for this provider. Make sure it does not clash with the keys of other providers.
    stateKey: this.GetType().Name,
    // An optional jsonSerializerOptions to control the serialization/deserialization of the custom state object
    jsonSerializerOptions: myJsonSerializerOptions);

// Using the helper you can read state:
MyCustomState state = sessionStateHelper.GetOrInitializeState(session);
Console.WriteLine(state.MemoryId);

// And write state:
sessionStateHelper.SaveState(session, state);

Jednoduchá AIContextProvider implementace

Nejjednodušší AIContextProvider implementace by obvykle přepsala dvě metody:

• AIContextProvider.ProvideAIContextAsync – Načtěte relevantní data a vraťte další pokyny, zprávy nebo nástroje.
• AIContextProvider.StoreAIContextAsync – extrahuje všechna relevantní data z nových zpráv a úložiště.

Tady je příklad jednoduché AIContextProvider integrace se službou paměti.

internal sealed class SimpleServiceMemoryProvider : AIContextProvider
{
    private readonly ProviderSessionState<State> _sessionState;
    private readonly ServiceClient _client;

    public SimpleServiceMemoryProvider(ServiceClient client, Func<AgentSession?, State>? stateInitializer = null)
        : base(null, null)
    {
        this._sessionState = new ProviderSessionState<State>(
            stateInitializer ?? (_ => new State()),
            this.GetType().Name);
        this._client = client;
    }

    public override string StateKey => this._sessionState.StateKey;

    protected override ValueTask<AIContext> ProvideAIContextAsync(InvokingContext context, CancellationToken cancellationToken = default)
    {
        var state = this._sessionState.GetOrInitializeState(context.Session);

        if (state.MemoriesId == null)
        {
            // No stored memories yet.
            return new ValueTask<AIContext>(new AIContext());
        }

        // Find memories that match the current user input.
        var memories = this._client.LoadMemories(state.MemoriesId, string.Join("\n", context.AIContext.Messages?.Select(x => x.Text) ?? []));

        // Return a new message that contains the text from any memories that were found.
        return new ValueTask<AIContext>(new AIContext
        {
            Messages = [new ChatMessage(ChatRole.User, "Here are some memories to help answer the user question: " + string.Join("\n", memories.Select(x => x.Text)))]
        });
    }

    protected override async ValueTask StoreAIContextAsync(InvokedContext context, CancellationToken cancellationToken = default)
    {
        var state = this._sessionState.GetOrInitializeState(context.Session);
        // Create a memory container in the service for this session
        // and save the returned id in the session.
        state.MemoriesId ??= this._client.CreateMemoryContainer();
        this._sessionState.SaveState(context.Session, state);

        // Use the service to extract memories from the user input and agent response.
        await this._client.StoreMemoriesAsync(state.MemoriesId, context.RequestMessages.Concat(context.ResponseMessages ?? []), cancellationToken);
    }

    public class State
    {
        public string? MemoriesId { get; set; }
    }
}

Pokročilá AIContextProvider implementace

Pokročilejší implementace by se mohla rozhodnout přepsat následující metody:

• AIContextProvider.InvokingCoreAsync – Volá se před vyvoláním LLM a umožňuje provést úpravu seznamu zpráv požadavků, nástrojů a pokynů.
• AIContextProvider.InvokedCoreAsync – volá se po vyvolání LLM agentem a umožňuje přístup ke všem zprávám požadavků a odpovědí.

AIContextProvider poskytuje základní implementace InvokingCoreAsync a InvokedCoreAsync.

Základní InvokingCoreAsync implementace provede následující:

• filtruje vstupní seznam zpráv pouze na zprávy, které byly předány agentovi volajícím. Všimněte si, že tento filtr lze přepsat pomocí provideInputMessageFilter parametru v konstruktoru AIContextProvider .
• volání ProvideAIContextAsync s filtrovanými požadovanými zprávami, existujícími nástroji a pokyny.
• označí všechny zprávy vrácené zdrojovými ProvideAIContextAsync informacemi, které označují, že tyto zprávy pocházejí od tohoto zprostředkovatele kontextu.
• sloučí zprávy, nástroje a pokyny vrácené z ProvideAIContextAsync s těmi stávajícími, aby se vytvořil vstup, který agent použije. Zprávy, nástroje a pokyny se připojují k existujícím.

Základ InvokedCoreAsync dělá následující:

• zkontroluje, jestli spuštění selhalo, a pokud ano, vrátí se bez dalšího zpracování.
• filtruje vstupní seznam zpráv pouze na zprávy, které byly předány agentovi volajícím. Všimněte si, že tento filtr lze přepsat pomocí storeInputMessageFilter parametru v konstruktoru AIContextProvider .
• předá filtrované požadované zprávy a všechny zprávy odpovědi do StoreAIContextAsync úložiště.

Tyto metody je možné přepsat k implementaci AIContextProvider, ale to vyžaduje, aby implementátor sám přiměřeně zajistil základní funkčnost. Tady je příklad takové implementace.

internal sealed class AdvancedServiceMemoryProvider : AIContextProvider
{
    private readonly ProviderSessionState<State> _sessionState;
    private readonly ServiceClient _client;

    public AdvancedServiceMemoryProvider(ServiceClient client, Func<AgentSession?, State>? stateInitializer = null)
        : base(null, null)
    {
        this._sessionState = new ProviderSessionState<State>(
            stateInitializer ?? (_ => new State()),
            this.GetType().Name);
        this._client = client;
    }

    public override string StateKey => this._sessionState.StateKey;

    protected override async ValueTask<AIContext> InvokingCoreAsync(InvokingContext context, CancellationToken cancellationToken = default)
    {
        var state = this._sessionState.GetOrInitializeState(context.Session);

        if (state.MemoriesId == null)
        {
            // No stored memories yet.
            return new AIContext();
        }

        // We only want to search for memories based on user input, and exclude chat history or other AI context provider messages.
        var filteredInputMessages = context.AIContext.Messages?.Where(m => m.GetAgentRequestMessageSourceType() == AgentRequestMessageSourceType.External);

        // Find memories that match the current user input.
        var memories = this._client.LoadMemories(state.MemoriesId, string.Join("\n", filteredInputMessages?.Select(x => x.Text) ?? []));

        // Create a message for the memories, and stamp it to indicate where it came from.
        var memoryMessages =
            [new ChatMessage(ChatRole.User, "Here are some memories to help answer the user question: " + string.Join("\n", memories.Select(x => x.Text)))]
            .Select(m => m.WithAgentRequestMessageSource(AgentRequestMessageSourceType.AIContextProvider, this.GetType().FullName!));

        // Return a new merged AIContext.
        return new AIContext
        {
            Instructions = context.AIContext.Instructions,
            Messages = context.AIContext.Messages.Concat(memoryMessages),
            Tools = context.AIContext.Tools
        };
    }

    protected override async ValueTask InvokedCoreAsync(InvokedContext context, CancellationToken cancellationToken = default)
    {
        if (context.InvokeException is not null)
        {
            return;
        }

        var state = this._sessionState.GetOrInitializeState(context.Session);
        // Create a memory container in the service for this session
        // and save the returned id in the session.
        state.MemoriesId ??= this._client.CreateMemoryContainer();
        this._sessionState.SaveState(context.Session, state);

        // We only want to store memories based on user input and agent output, and exclude messages from chat history or other AI context providers to avoid feedback loops.
        var filteredRequestMessages = context.RequestMessages.Where(m => m.GetAgentRequestMessageSourceType() == AgentRequestMessageSourceType.External);

        // Use the service to extract memories from the user input and agent response.
        await this._client.StoreMemoriesAsync(state.MemoriesId, filteredRequestMessages.Concat(context.ResponseMessages ?? []), cancellationToken);
    }

    public class State
    {
        public string? MemoriesId { get; set; }
    }
}

from typing import Any

from agent_framework import AgentSession, ContextProvider, SessionContext


class UserPreferenceProvider(ContextProvider):
    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(".")

Poznámka:

ContextProvider a HistoryProvider jsou kanonické Python základní třídy. BaseContextProvider a BaseHistoryProvider stále existují jako zastaralé aliasy kvůli kompatibilitě, ale noví zprostředkovatelé by měli dědit z nových názvů.

Poskytovatelé kontextu mohou také přidat chatové nebo funkční middleware pro aktuální volání pomocí context.extend_middleware(self.source_id, middleware). Agent tyto doplňky context.get_middleware() slučuje a použije je v daném pořadí poskytovatelů před spuštěním chatovacího klienta.

Dynamický výběr nástrojů se sadami nástrojů Foundry

Zprostředkovatelé kontextu můžou dynamicky přidávat nebo odebírat nástroje při každém zapnutí. Příklad, který používá sadu nástrojů Foundry k výběru nástrojů v každé interakci podle zprávy uživatele, najdete v ukázce foundry_toolbox_context_provider. Další informace o panelech nástrojů najdete v tématu Sady nástrojů Foundry.

Vlastní zprostředkovatel historie

Zprostředkovatelé historie jsou zprostředkovatelé kontextu specializovaní na načítání a ukládání zpráv.

from collections.abc import Sequence
from typing import Any

from agent_framework import HistoryProvider, Message


class DatabaseHistoryProvider(HistoryProvider):
    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])

Důležité

V Pythonu můžete nakonfigurovat více poskytovatelů historie, ale pouze jeden by měl používat load_messages=True. Používejte další zprostředkovatele pro diagnostiku a hodnocení spolu s load_messages=False a store_context_messages=True, tak aby zachytili kontext od jiných poskytovatelů společně se vstupem a výstupem. Pokud potřebujete, aby místní historie přetrvávala kolem každého volání modelu ve smyčce nástroje, přečtěte si Úložiště.

Příklad vzoru:

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

Další kroky

Úložiště