Rasa Pro Version Migration Guide

a2a_server:
  url: "https://rasa.example.com"
  description: "My domain sub-agent"

session_config:
  session_expiration_time: 60
  start_session_after_expiry: false

SANIC_WORKERS=1 rasa run -m models/your-model.tar.gz --endpoints endpoints.yml

# A2A agents
async def process_agent_output(self, output: AgentOutput) -> AgentOutput

# ReAct / MCP agents
async def process_tool_output(
    self,
    current_iteration_tool_results: Dict[str, AgentToolResult],
    cumulative_tool_results: Dict[str, AgentToolResult],
    output_channel: Optional[OutputChannel] = None,
) -> List[Event]

async def run(
    self,
    input: "AgentInput",
    output_channel: Optional[OutputChannel] = None,
    cancellation_token: Optional["CancellationToken"] = None,
) -> "AgentOutput"

from rasa.agents.schemas import AgentToolContext

async def retrieve(self, sender_id: Text) -> Optional[DialogueStateTracker]:
    # ... retrieve events and user_id from storage ...
    
    return DialogueStateTracker.from_dict(
        sender_id, 
        events, 
        slots,
        user_id=user_id  # Add this parameter
    )

class MyCustomTrackerStore(TrackerStore):
    def serialise_tracker(self, tracker: DialogueStateTracker) -> Any:
        # You must manually include user_id in your serialization
        return {
            "sender_id": tracker.sender_id,
            "user_id": tracker.user_id,  # Add this
            "events": [e.as_dict() for e in tracker.events],
            # ... other fields
        }

def deserialise_tracker(
    self, sender_id: Text, serialised_tracker: Any
) -> Optional[DialogueStateTracker]:
    # Extract user_id from your serialized format
    user_id = serialised_tracker.get("user_id")
    
    tracker = self.init_tracker(sender_id, user_id=user_id)
    # ... recreate tracker state ...
    return tracker

async def get_serialized_trackers_by_user_id(
    self,
    user_id: str,
    limit: Optional[int] = None,
    skip: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """Retrieve serialized trackers for the given user_id.

    Args:
        user_id: User ID to retrieve trackers for.
        limit: Optional limit on the number of trackers to return.
        skip: Optional offset for pagination.

    Returns:
        List of serialized tracker dicts for the given user_id, ordered by
        conversation_started_timestamp and sender_id, without replaying
        events through DialogueStateTracker.
    """
    # Fetch raw serialized tracker dicts from your storage
    serialized_trackers = []
    # ... your implementation to load serialized data by user_id ...

    # Return directly — no DialogueStateTracker instantiation needed
    return serialized_trackers

from rasa.shared.utils.llm import LLMInput

# Create LLMInput with prompt and metadata
llm_input = LLMInput(
    prompt=your_prompt,
    metadata=self.get_llm_tracing_metadata(tracker)
)

# Call invoke_llm with LLMInput
response = await self.invoke_llm(llm_input)

async def run(
    self, input: "AgentInput", output_channel: Optional[OutputChannel] = None
) -> "AgentOutput":
    """Send a message to Agent/server and return response.
    Args:
        input: The agent input containing user message and context
        output_channel: Optional output channel for sending intermediate messages
    Returns:
        The agent output containing response and status
    """

{% if current_datetime %}
---

### Date & Time Context
- Current date: {{ current_datetime.strftime("%d %B, %Y") }}   (DD Month, YYYY)
- Current time: {{ current_datetime.strftime("%H:%M:%S") }} ({{ current_datetime.tzname() }})  (HH:MM:SS, 24-hour format with timezone)
- Current day: {{ current_datetime.strftime("%A") }}     (Day of week)
{% endif %}

def completion(
    self, messages: Union[List[dict], List[str], str], **kwargs: Any
) -> LLMResponse:

async def acompletion(
    self, messages: Union[List[dict], List[str], str], **kwargs: Any
) -> LLMResponse:

@dataclass
class LLMResponse:
    id: str
    """A unique identifier for the completion."""

    choices: List[str]
    """The list of completion choices the model generated for the input prompt."""

    created: int
    """The Unix timestamp (in seconds) of when the completion was created."""

    model: Optional[str] = None
    """The model used for completion."""

    usage: Optional[LLMUsage] = None
    """An optional details about the token usage for the API call."""

    additional_info: Optional[Dict] = None
    """Optional dictionary for storing additional information related to the
    completion that may not be covered by other fields."""

    latency: Optional[float] = None
    """Optional field to store the latency of the LLM API call."""

    tool_calls: Optional[List[LLMToolCall]] = None
    """The list of tool calls the model generated for the input prompt."""

class LLMToolCall(BaseModel):
    """A class representing a response from an LLM tool call."""

    id: str
    """The ID of the tool call."""

    tool_name: str
    """The name of the tool that was called."""

    tool_args: Dict[str, Any]
    """The arguments passed to the tool call."""

    type: str = "function"
    """The type of the tool call."""

    def run_command_on_tracker(
        self,
        tracker: DialogueStateTracker,
        all_flows: FlowsList,
        original_tracker: DialogueStateTracker,
    ) -> List[Event]:
        """Runs the command on the tracker.

        Args:
            tracker: The tracker to run the command on.
            all_flows: All flows in the assistant.
            original_tracker: The tracker before any command was executed.

        Returns:
            The events to apply to the tracker.
        """
        stack = tracker.stack
        original_stack = original_tracker.stack
        applied_events: List[Event] = []

        if self.flow not in all_flows.flow_ids:
            structlogger.debug(
                "start_flow_command.skip_command.start_invalid_flow_id", command=self
            )
            return []

        original_user_frame = top_user_flow_frame(original_stack)
        original_top_flow = (
            original_user_frame.flow(all_flows) if original_user_frame else None
        )

        # if the original top flow is the same as the flow to start, the flow is
        # already active, do nothing
        if original_top_flow is not None and original_top_flow.id == self.flow:
            # in case continue_interrupted is not active, skip the already active start
            # flow command
            if not is_continue_interrupted_flow_active(stack):
                return []

            # if the continue interrupted flow is active, and the command generator
            # predicted a start flow command for the flow which is on top of the stack,
            # we just need to remove the pattern_continue_interrupted frame(s) from the
            # stack
            stack, flow_completed_events = remove_pattern_continue_interrupted_frames(
                stack
            )
            applied_events.extend(flow_completed_events)
            return applied_events + tracker.create_stack_updated_events(stack)

        # if the flow is already on the stack, resume it
        if (
            self.flow in user_flows_on_the_stack(stack)
            and original_user_frame is not None
        ):
            # if pattern_continue_interrupted is active, we need to remove it
            # from the stack before resuming the flow
            stack, flow_completed_events = remove_pattern_continue_interrupted_frames(
                stack
            )
            applied_events.extend(flow_completed_events)
            applied_events.extend(resume_flow(self.flow, tracker, stack))
            # the current active flow is interrupted
            applied_events.append(
                FlowInterrupted(
                    original_user_frame.flow_id, original_user_frame.step_id
                )
            )
            return applied_events

        frame_type = FlowStackFrameType.REGULAR

        # remove the pattern_continue_interrupted frames from the stack
        # if it is currently active but the user digressed from the pattern
        stack, flow_completed_events = remove_pattern_continue_interrupted_frames(stack)
        applied_events.extend(flow_completed_events)

        if original_top_flow:
            # if the original top flow is not the same as the flow to start,
            # interrupt the current active flow
            frame_type = FlowStackFrameType.INTERRUPT

            if original_user_frame is not None:
                applied_events.append(
                    FlowInterrupted(
                        original_user_frame.flow_id, original_user_frame.step_id
                    )
                )

            # If there is an active agent frame, interrupt it
            active_agent_stack_frame = stack.find_active_agent_frame()
            if active_agent_stack_frame:
                structlogger.debug(
                    "start_flow_command.interrupt_agent",
                    command=self,
                    agent_id=active_agent_stack_frame.agent_id,
                    frame_id=active_agent_stack_frame.frame_id,
                    flow_id=active_agent_stack_frame.flow_id,
                )
                active_agent_stack_frame.state = AgentState.INTERRUPTED
                applied_events.append(
                    AgentInterrupted(
                        active_agent_stack_frame.agent_id,
                        active_agent_stack_frame.flow_id,
                    )
                )

        structlogger.debug("start_flow_command.start_flow", command=self)
        stack.push(UserFlowStackFrame(flow_id=self.flow, frame_type=frame_type))
        return applied_events + tracker.create_stack_updated_events(stack)

def run_command_on_tracker(
    self,
    tracker: DialogueStateTracker,
    all_flows: FlowsList,
    original_tracker: DialogueStateTracker,
) -> List[Event]:
    """Runs the command on the tracker.

    Args:
        tracker: The tracker to run the command on.
        all_flows: All flows in the assistant.
        original_tracker: The tracker before any command was executed.

    Returns:
        The events to apply to the tracker.
    """
    stack = tracker.stack
    original_stack = original_tracker.stack

    applied_events: List[Event] = []
    user_frame = top_user_flow_frame(
        original_stack, ignore_call_and_link_frames=False
    )
    current_flow = user_frame.flow(all_flows) if user_frame else None

    if not current_flow:
        structlogger.debug(
            "cancel_command.skip_cancel_flow.no_active_flow", command=self
        )
        return []

    if agent_frame := original_tracker.stack.find_active_agent_stack_frame_for_flow(
        current_flow.id
    ):
        structlogger.debug(
            "cancel_command.remove_agent_stack_frame",
            command=self,
            frame=agent_frame,
        )
        remove_agent_stack_frame(stack, agent_frame.agent_id)
        applied_events.append(
            AgentCancelled(agent_id=agent_frame.agent_id, flow_id=current_flow.id)
        )

    # we pass in the original dialogue stack (before any of the currently
    # predicted commands were applied) to make sure we don't cancel any
    # frames that were added by the currently predicted commands.
    canceled_frames = self.select_canceled_frames(original_stack)

    stack.push(
        CancelPatternFlowStackFrame(
            canceled_name=current_flow.readable_name(
                language=tracker.current_language
            ),
            canceled_frames=canceled_frames,
        )
    )

    if user_frame:
        applied_events.append(FlowCancelled(user_frame.flow_id, user_frame.step_id))

    return applied_events + tracker.create_stack_updated_events(stack)

def run_command_on_tracker(
    self,
    tracker: DialogueStateTracker,
    all_flows: FlowsList,
    original_tracker: DialogueStateTracker,
) -> List[Event]:
    """Runs the command on the tracker.

    Args:
        tracker: The tracker to run the command on.
        all_flows: All flows in the assistant.
        original_tracker: The tracker before any command was executed.

    Returns:
        The events to apply to the tracker.
    """
    flows = [all_flows.flow_by_id(opt) for opt in self.options]
    clean_options = [flow.id for flow in flows if flow is not None]
    if len(clean_options) != len(self.options):
        structlogger.debug(
            "clarify_command.altered_command.dropped_clarification_options",
            command=self,
            original_options=self.options,
            cleaned_options=clean_options,
        )
    if len(clean_options) == 0:
        structlogger.debug(
            "clarify_command.skip_command.empty_clarification", command=self
        )
        return []

    stack = tracker.stack
    relevant_flows = [all_flows.flow_by_id(opt) for opt in clean_options]

    names = [
        flow.readable_name(language=tracker.current_language)
        for flow in relevant_flows
        if flow is not None
    ]

    applied_events: List[Event] = []

    # if the top stack frame is an agent stack frame, we need to
    # update the state to INTERRUPTED and add an AgentInterrupted event
    if top_stack_frame := stack.top():
        if isinstance(top_stack_frame, AgentStackFrame):
            applied_events.append(
                AgentInterrupted(
                    top_stack_frame.agent_id,
                    top_stack_frame.flow_id,
                )
            )
            top_stack_frame.state = AgentState.INTERRUPTED

    stack.push(ClarifyPatternFlowStackFrame(names=names))
    return applied_events + tracker.create_stack_updated_events(stack)

def run_command_on_tracker(
    self,
    tracker: DialogueStateTracker,
    all_flows: FlowsList,
    original_tracker: DialogueStateTracker,
) -> List[Event]:
    """Runs the command on the tracker.

    Args:
        tracker: The tracker to run the command on.
        all_flows: All flows in the assistant.
        original_tracker: The tracker before any command was executed.

    Returns:
        The events to apply to the tracker.
    """
    stack = tracker.stack

    applied_events: List[Event] = []

    # if the top stack frame is an agent stack frame, we need to
    # update the state to INTERRUPTED and add an AgentInterrupted event
    if top_stack_frame := stack.top():
        if isinstance(top_stack_frame, AgentStackFrame):
            applied_events.append(
                AgentInterrupted(
                    top_stack_frame.agent_id,
                    top_stack_frame.flow_id,
                )
            )
            top_stack_frame.state = AgentState.INTERRUPTED

    stack.push(ChitchatPatternFlowStackFrame())
    return applied_events + tracker.create_stack_updated_events(stack)

def run_command_on_tracker(
    self,
    tracker: DialogueStateTracker,
    all_flows: FlowsList,
    original_tracker: DialogueStateTracker,
) -> List[Event]:
    """Runs the command on the tracker.

    Args:
        tracker: The tracker to run the command on.
        all_flows: All flows in the assistant.
        original_tracker: The tracker before any command was executed.

    Returns:
        The events to apply to the tracker.
    """
    stack = tracker.stack

    applied_events: List[Event] = []

    # if the top stack frame is an agent stack frame, we need to
    # update the state to INTERRUPTED and add an AgentInterrupted event
    if top_stack_frame := stack.top():
        if isinstance(top_stack_frame, AgentStackFrame):
            applied_events.append(
                AgentInterrupted(
                    top_stack_frame.agent_id,
                    top_stack_frame.flow_id,
                )
            )
            top_stack_frame.state = AgentState.INTERRUPTED

    stack.push(SearchPatternFlowStackFrame())
    return applied_events + tracker.create_stack_updated_events(stack)

docker run --rm --name jaeger \
  -p 16686:16686 \
  -p 4317:4317 \
  -p 4318:4318 \
  -p 5778:5778 \
  -p 9411:9411 \
  cr.jaegertracing.io/jaegertracing/jaeger:2.10.0

tracing:
  type: jaeger
  host: 0.0.0.0
  port: 4317
  service_name: rasa
  sync_export: ~

prior_commands = self._get_prior_commands(tracker)
if tracker is None or flows.is_empty():
    return prior_commands

# custom command generation logic block

return self._check_commands_overlap(prior_commands, commands)

# Old class definition:
from rasa.dialogue_understanding.generators import SingleStepLLMCommandGenerator

class MyCommandGenerator(SingleStepLLMCommandGenerator):
    ...

# New class definition:
from rasa.dialogue_understanding.generators import CompactLLMCommandGenerator

class MyCommandGenerator(CompactLLMCommandGenerator):
    ...

"Ok, I am updating {{ context.corrected_slots.keys()|join(', ') }} to {{ context.new_slot_values | join(', ') }} respectively."

llm_judge:
  llm:
    provider: "openai"
    model: "gpt-4-0613"
  embeddings:
    provider: "openai"
    model: "text-embedding-ada-002"

llm:
  provider: "cohere" # instead of "type: cohere"
  model: "command-r"

llm:
  provider: "vertex_ai" # instead of "type: vertexai"
  model: "gemini-pro"

llm:
  provider: "huggingface" # instead of "type: huggingface_hub"
  model: "HuggingFaceH4/zephyr-7b-beta" # instead of "repo_id: HuggingFaceH4/zephyr-7b-beta"

llm:
  provider: "self-hosted" # instead of "type: llamacpp"
  api_base: "http://localhost:8000/v1" # instead of "model_path: "/path/to/model.bin""
  model: "ggml-org/Meta-Llama-3.1-8B-Instruct-Q4_0-GGUF"

llm:
  provider: "self-hosted" # instead of "type: vllm_openai"
  api_base: "http://localhost:8000/v1"
  model: "NousResearch/Meta-Llama-3-8B-Instruct"  # the name of the model you have deployed

llm:
    model: ...
    cache: false

DEFAULT_LLM_CONFIG = {
    "_type": "openai",
    "model_name": ...,
    "request_timeout": ...,
    "temperature": ...,
    "max_tokens": ...,
}

DEFAULT_LLM_CONFIG = {
    "provider": "openai",
    "model": ...,
    "temperature": ...,
    "max_tokens": ...,
    "timeout": ...,
}

DEFAULT_EMBEDDINGS_CONFIG = {
    "_type": "openai",
    "model": ...,
}