ContextProvider Class

Base class for all context providers.

A context provider is a component that can be used to enhance the AI's context management. It can listen to changes in the conversation and provide additional context to the AI model just before invocation.

Note

ContextProvider is an abstract base class. You must subclass it and implement

the invoking() method to create a custom context provider. Ideally, you should

also implement the invoked() and thread_created() methods to track conversation

state, but these are optional.

Constructor

ContextProvider()

Examples

   from agent_framework import ContextProvider, Context, ChatMessage


   class CustomContextProvider(ContextProvider):
       async def invoking(self, messages, **kwargs):
           # Add custom instructions before each invocation
           return Context(instructions="Always be concise and helpful.", messages=[], tools=[])


   # Use with a chat agent
   async with CustomContextProvider() as provider:
       agent = ChatAgent(chat_client=client, name="assistant", context_providers=provider)

Methods

__init__
__new__
invoked	Called after the agent has received a response from the underlying inference service. You can inspect the request and response messages, and update the state of the context provider.
invoking	Called just before the model/agent is invoked. Implementers can load any additional context required at this time, and they should return any context that should be passed to the agent.
thread_created	Called just after a new thread is created. Implementers can use this method to perform any operations required at the creation of a new thread. For example, checking long-term storage for any data that is relevant to the current session.

__init__

__init__()

__new__

__new__(**kwargs)

invoked

Called after the agent has received a response from the underlying inference service.

You can inspect the request and response messages, and update the state of the context provider.

async invoked(request_messages: ChatMessage | Sequence[ChatMessage], response_messages: ChatMessage | Sequence[ChatMessage] | None = None, invoke_exception: Exception | None = None, **kwargs: Any) -> None

Parameters

Name	Description
request_messages Required	ChatMessage | Sequence[ChatMessage] The messages that were sent to the model/agent.
response_messages	ChatMessage | Sequence[ChatMessage] | None The messages that were returned by the model/agent. Default value: None
invoke_exception	Exception | None The exception that was thrown, if any. Default value: None
kwargs Required	Any

Keyword-Only Parameters

Name	Description
kwargs	Additional keyword arguments (not used at present).

Returns

Type	Description
None

invoking

Called just before the model/agent is invoked.

Implementers can load any additional context required at this time, and they should return any context that should be passed to the agent.

abstract async invoking(messages: ChatMessage | MutableSequence[ChatMessage], **kwargs: Any) -> Context

Parameters

Name	Description
messages Required	ChatMessage | MutableSequence[ChatMessage] The most recent messages that the agent is being invoked with.
kwargs Required	Any

Keyword-Only Parameters

Name	Description
kwargs	Additional keyword arguments (not used at present).

Returns

Type	Description
Context	A Context object containing instructions, messages, and tools to include.

thread_created

Called just after a new thread is created.

Implementers can use this method to perform any operations required at the creation of a new thread. For example, checking long-term storage for any data that is relevant to the current session.

async thread_created(thread_id: str | None) -> None

Parameters

Name	Description
thread_id Required	str | None The ID of the new thread.

Returns

Type	Description
None

Attributes

DEFAULT_CONTEXT_PROMPT

DEFAULT_CONTEXT_PROMPT: Final[str] = '## Memories\nConsider the following memories when answering user questions:'