bayesianbandits.AgentPipeline#

bayesianbandits.AgentPipeline(steps: List[Tuple[str, Any]], final_agent: ContextualAgent[ContextType, TokenType]) → ContextualAgentPipeline[ContextType, TokenType]#
bayesianbandits.AgentPipeline(steps: List[Tuple[str, Any]], final_agent: Agent[TokenType]) → NonContextualAgentPipeline[TokenType]

Create a Pipeline that wraps an Agent or ContextualAgent.

This factory function provides a clean API for creating pipelines while maintaining complete static typing based on the agent type. The pipeline can accept any input type and transform it to what the agent expects.

The resulting Pipeline will have the same interface as the wrapped agent, allowing you to call pull, update, and other methods directly on it.

Parameters:

  • steps (List[Tuple[str, Any]]) – List of (name, transformer) tuples for preprocessing steps. All transformers must be either stateless or pre-fitted. The output of the transformation chain must match the agent’s expected input type.

  • final_agent (Agent[TokenType] or ContextualAgent[ContextType, TokenType]) – The agent to wrap. The pipeline type is determined by the agent type.

Returns:

The appropriate pipeline type based on the final_agent type.

Return type:

ContextualAgentPipeline or NonContextualAgentPipeline

Examples

>>> from sklearn.feature_extraction import DictVectorizer
>>> from bayesianbandits import Arm, NormalRegressor, ContextualAgent, ThompsonSampling
>>>
>>> # Pipeline accepting dict input, outputting sparse arrays for agent
>>> arms = [Arm(i, learner=NormalRegressor(alpha=1.0, beta=1.0, sparse=True)) for i in range(3)]
>>> agent = ContextualAgent(arms, ThompsonSampling())
>>> vectorizer = DictVectorizer()
>>> _ = vectorizer.fit([{'user': 'A'}, {'user': 'B'}])
>>>
>>> pipeline = AgentPipeline(
...     steps=[('vectorize', vectorizer)],
...     final_agent=agent
... )
>>> # Can accept dict input: [{'user': 'A', 'item': 1}]
>>> # Transforms to sparse matrix for agent
On this page

This Page