feat(experimental): add /btw commands to steer conversational flows

Opt-in add-on that intercepts /btw lines before handle_turn so users can inject steering notes or force a route without recording a user turn. Co-authored-by: Lorenze Jay <lorenzejay@users.noreply.github.com>
2026-09-22 10:56:50 +00:00 · 2026-08-31 06:43:59 +00:00
parent da4daadba0
commit 69e7bc01f4
10 changed files with 903 additions and 0 deletions
--- a/docs/edge/ar/guides/flows/conversational-flows.mdx
+++ b/docs/edge/ar/guides/flows/conversational-flows.mdx
@@ -612,6 +612,44 @@ from crewai.flow import (
 )
 ```

+## تجريبي: أوامر `/btw`
+
+<Warning>
+`/btw` إضافة تجريبية. تتجاهل التدفقات المحادثية أسطر الأوامر حتى تفعّلها صراحة، وقد يتغيّر الـ API.
+</Warning>
+
+يعامل كل `handle_turn()` سطر المستخدم كتشغيل جديد للرسم. `/btw` قناة جانبية تتيح لك إدخال ملاحظة توجيه — أو فرض مسار — **دون** إلحاق رسالة مستخدم أو تشغيل الرسم.
+
+```python
+from crewai.experimental.conversation_commands import btw_commands
+from crewai.flow import ConversationConfig, ConversationState, Flow, listen
+
+
+@btw_commands
+@ConversationConfig()
+class SupportFlow(Flow[ConversationState]):
+    @listen("RESEARCH")
+    def handle_research(self) -> str:
+        """Fresh web research."""
+        reply = "I would research that."
+        self.append_assistant_message(reply)
+        return reply
+```
+
+| السطر | الأثر |
+|------|--------|
+| `/btw keep answers under 20 words` | يحفظ ملاحظة توجيه. لا تُسجَّل جولة مستخدم. تظهر الملاحظة لاحقاً في مطالبات `converse` والموجّه. |
+| `/btw route RESEARCH` | يفرض المسار على **الجولة التالية** ثم يعود للتوجيه العادي. |
+| `/btw stay RESEARCH` أو `/btw route RESEARCH persist` | يواصل فرض ذلك المسار حتى `/btw clear`. |
+| `/btw clear` | يحذف الملاحظات والمسارات المفروضة. |
+| `/btw` أو `/btw show` | يعرض التوجيه الحالي. |
+| `/help` | يسرد الأوامر. |
+| `What's the weather /btw keep it to one sentence` | يطبّق الأمر ثم يشغّل النص المتبقي كجولة مستخدم. |
+
+لتدفق أنشأته مسبقاً (بما في ذلك `Flow.from_declaration(...)`)، استدعِ `enable_btw_commands(flow)` بدل المزيّن.
+
+أسطر `/btw` المستقلة تعيد إقراراً ولا تستدعي `kickoff()`. ولأن `chat()` وواجهة CLI TUI تغلفان `handle_turn()`، تلتقطان الإضافة تلقائياً بعد تثبيتها. التدفقات بدون الإضافة ما زالت تعامل `/btw …` كرسالة مستخدم عادية.
+
 ## مراجع

 - [إتقان إدارة حالة Flow](/ar/guides/flows/mastering-flow-state)
--- a/docs/edge/en/guides/flows/conversational-flows.mdx
+++ b/docs/edge/en/guides/flows/conversational-flows.mdx
@@ -613,6 +613,44 @@ from crewai.flow import (
 )
 ```

+## Experimental: `/btw` commands
+
+<Warning>
+`/btw` is an experimental add-on. Conversational flows ignore slash lines until you opt in, and the API may change.
+</Warning>
+
+Each `handle_turn()` treats the user line as a new graph run. `/btw` is a side-channel so you can interject a steering note — or force a route — **without** appending a user message or running the graph.
+
+```python
+from crewai.experimental.conversation_commands import btw_commands
+from crewai.flow import ConversationConfig, ConversationState, Flow, listen
+
+
+@btw_commands
+@ConversationConfig()
+class SupportFlow(Flow[ConversationState]):
+    @listen("RESEARCH")
+    def handle_research(self) -> str:
+        """Fresh web research."""
+        reply = "I would research that."
+        self.append_assistant_message(reply)
+        return reply
+```
+
+| Line | Effect |
+|------|--------|
+| `/btw keep answers under 20 words` | Persist a steering note. No user turn is recorded. Later `converse` and router prompts include the note. |
+| `/btw route RESEARCH` | Force the **next** turn onto that route, then resume normal routing. |
+| `/btw stay RESEARCH` or `/btw route RESEARCH persist` | Keep forcing that route until `/btw clear`. |
+| `/btw clear` | Drop notes and forced routes. |
+| `/btw` or `/btw show` | Print the current steering. |
+| `/help` | List commands. |
+| `What's the weather /btw keep it to one sentence` | Apply the command, then run the leftover text as the user turn. |
+
+For a flow you already constructed (including `Flow.from_declaration(...)`), call `enable_btw_commands(flow)` instead of the decorator.
+
+Standalone `/btw` lines return an acknowledgement and do not call `kickoff()`. Because `chat()` and the CLI TUI wrap `handle_turn()`, they pick the add-on up automatically once it is installed. Flows without the add-on still treat `/btw …` as a normal user message.
+
 ## See also

 - [Mastering Flow State Management](/en/guides/flows/mastering-flow-state) — persistence, Pydantic state, `@persist`
--- a/docs/edge/ko/guides/flows/conversational-flows.mdx
+++ b/docs/edge/ko/guides/flows/conversational-flows.mdx
@@ -608,6 +608,44 @@ from crewai.flow import (
 )
 ```

+## 실험적: `/btw` 명령
+
+<Warning>
+`/btw`는 실험적 애드온입니다. 대화형 Flow는 명시적으로 켜기 전까지 슬래시 줄을 무시하며, API는 바뀔 수 있습니다.
+</Warning>
+
+각 `handle_turn()`은 사용자 줄을 새로운 그래프 실행으로 취급합니다. `/btw`는 사이드 채널이므로 사용자 메시지를 추가하거나 그래프를 실행하지 **않고** 스티어링 노트(또는 강제 라우트)를 끼워 넣을 수 있습니다.
+
+```python
+from crewai.experimental.conversation_commands import btw_commands
+from crewai.flow import ConversationConfig, ConversationState, Flow, listen
+
+
+@btw_commands
+@ConversationConfig()
+class SupportFlow(Flow[ConversationState]):
+    @listen("RESEARCH")
+    def handle_research(self) -> str:
+        """Fresh web research."""
+        reply = "I would research that."
+        self.append_assistant_message(reply)
+        return reply
+```
+
+| 줄 | 효과 |
+|------|--------|
+| `/btw keep answers under 20 words` | 스티어링 노트를 유지합니다. 사용자 턴은 기록되지 않습니다. 이후 `converse`와 라우터 프롬프트에 노트가 포함됩니다. |
+| `/btw route RESEARCH` | **다음** 턴만 해당 라우트로 강제하고 이후에는 일반 라우팅으로 돌아갑니다. |
+| `/btw stay RESEARCH` 또는 `/btw route RESEARCH persist` | `/btw clear` 전까지 그 라우트를 계속 강제합니다. |
+| `/btw clear` | 노트와 강제 라우트를 제거합니다. |
+| `/btw` 또는 `/btw show` | 현재 스티어링을 보여 줍니다. |
+| `/help` | 명령 목록을 출력합니다. |
+| `What's the weather /btw keep it to one sentence` | 명령을 적용한 뒤 남은 텍스트를 사용자 턴으로 실행합니다. |
+
+이미 만든 Flow( `Flow.from_declaration(...)` 포함)에는 데코레이터 대신 `enable_btw_commands(flow)`를 호출하세요.
+
+단독 `/btw` 줄은 확인 메시지를 반환하며 `kickoff()`를 호출하지 않습니다. `chat()`과 CLI TUI는 `handle_turn()`을 감싸므로, 애드온이 설치된 뒤에는 자동으로 동작합니다. 애드온이 없는 Flow는 `/btw …`를 일반 사용자 메시지로 취급합니다.
+
 ## 참고

 - [Flow 상태 관리 마스터하기](/ko/guides/flows/mastering-flow-state)
--- a/docs/edge/pt-BR/guides/flows/conversational-flows.mdx
+++ b/docs/edge/pt-BR/guides/flows/conversational-flows.mdx
@@ -613,6 +613,44 @@ from crewai.flow import (
 )
 ```

+## Experimental: comandos `/btw`
+
+<Warning>
+`/btw` é um extra experimental. Flows conversacionais ignoram linhas com barra até você ativar o extra, e a API pode mudar.
+</Warning>
+
+Cada `handle_turn()` trata a linha do usuário como uma nova execução do grafo. `/btw` é um canal lateral para intercalar uma nota de direção — ou forçar uma rota — **sem** anexar uma mensagem de usuário nem rodar o grafo.
+
+```python
+from crewai.experimental.conversation_commands import btw_commands
+from crewai.flow import ConversationConfig, ConversationState, Flow, listen
+
+
+@btw_commands
+@ConversationConfig()
+class SupportFlow(Flow[ConversationState]):
+    @listen("RESEARCH")
+    def handle_research(self) -> str:
+        """Fresh web research."""
+        reply = "I would research that."
+        self.append_assistant_message(reply)
+        return reply
+```
+
+| Linha | Efeito |
+|------|--------|
+| `/btw keep answers under 20 words` | Persiste uma nota de direção. Nenhum turno de usuário é gravado. As notas entram depois nos prompts de `converse` e do router. |
+| `/btw route RESEARCH` | Força o **próximo** turno para essa rota e depois volta ao roteamento normal. |
+| `/btw stay RESEARCH` ou `/btw route RESEARCH persist` | Continua forçando essa rota até `/btw clear`. |
+| `/btw clear` | Remove notas e rotas forçadas. |
+| `/btw` ou `/btw show` | Mostra a direção atual. |
+| `/help` | Lista os comandos. |
+| `What's the weather /btw keep it to one sentence` | Aplica o comando e executa o texto restante como turno do usuário. |
+
+Para um flow já construído (incluindo `Flow.from_declaration(...)`), chame `enable_btw_commands(flow)` em vez do decorator.
+
+Linhas `/btw` isoladas devolvem um acknowledgement e não chamam `kickoff()`. Como `chat()` e o TUI da CLI encapsulam `handle_turn()`, o extra passa a valer automaticamente depois de instalado. Flows sem o extra ainda tratam `/btw …` como mensagem normal de usuário.
+
 ## Veja também

 - [Dominando o Gerenciamento de Estado em Flows](/pt-BR/guides/flows/mastering-flow-state) — persistência, estado Pydantic, `@persist`
--- a/lib/crewai/src/crewai/experimental/init.py
+++ b/lib/crewai/src/crewai/experimental/init.py
@@ -20,6 +20,18 @@ from crewai.flow.conversational import (

 _LAZY_FROM_AGENT_EXECUTOR = {"AgentExecutor", "CrewAgentExecutorFlow"}

+_LAZY_FROM_CONVERSATION_COMMANDS = {
+    "BtwAction",
+    "BtwKind",
+    "BtwSteering",
+    "HELP_TEXT",
+    "ParsedBtwLine",
+    "btw_commands",
+    "enable_btw_commands",
+    "get_btw_steering",
+    "parse_btw_line",
+}
+
 _LAZY_FROM_EVALUATION = {
    "AgentEvaluationResult",
    "AgentEvaluator",
@@ -65,6 +77,13 @@ def __getattr__(name: str) -> Any:
            globals()[attr] = getattr(_evaluation_mod, attr)
        return globals()[name]

+    if name in _LAZY_FROM_CONVERSATION_COMMANDS:
+        from crewai.experimental import conversation_commands as _btw_mod
+
+        for attr in _LAZY_FROM_CONVERSATION_COMMANDS:
+            globals()[attr] = getattr(_btw_mod, attr)
+        return globals()[name]
+
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")


@@ -73,7 +92,12 @@ __all__ = [
    "AgentEvaluator",
    "AgentExecutor",
    "AgentMessage",
+    "BtwAction",
+    "BtwKind",
+    "BtwSteering",
    "BaseEvaluator",
+    "HELP_TEXT",
+    "ParsedBtwLine",
    "ConversationConfig",
    "ConversationEvent",
    "ConversationMessage",
@@ -92,6 +116,10 @@ __all__ = [
    "SemanticQualityEvaluator",
    "ToolInvocationEvaluator",
    "ToolSelectionEvaluator",
+    "btw_commands",
    "create_default_evaluator",
    "create_evaluation_callbacks",
+    "enable_btw_commands",
+    "get_btw_steering",
+    "parse_btw_line",
 ]
--- a/lib/crewai/src/crewai/experimental/conversation_commands/init.py
+++ b/lib/crewai/src/crewai/experimental/conversation_commands/init.py
@@ -0,0 +1,54 @@
+"""Experimental ``/btw`` commands for conversational Flows.
+
+Conversational turns treat every user line as a new graph run. This add-on
+lets you interject a side instruction that steers routing and replies
+without appending a user message. It is opt-in and not part of the stable
+``handle_turn`` / ``chat`` contract.
+
+Example::
+
+    from crewai.experimental.conversation_commands import btw_commands
+    from crewai.flow import ConversationConfig, ConversationState, Flow, listen
+
+
+    @btw_commands
+    @ConversationConfig()
+    class SupportFlow(Flow[ConversationState]):
+        @listen("RESEARCH")
+        def handle_research(self) -> str:
+            return "researching"
+
+    flow = SupportFlow()
+    flow.handle_turn("/btw keep answers under 20 words")
+    flow.handle_turn("/btw route RESEARCH")
+    flow.handle_turn("latest AI news")
+"""
+
+from crewai.experimental.conversation_commands.addon import (
+    btw_commands,
+    enable_btw_commands,
+)
+from crewai.experimental.conversation_commands.parser import (
+    HELP_TEXT,
+    BtwAction,
+    BtwKind,
+    ParsedBtwLine,
+    parse_btw_line,
+)
+from crewai.experimental.conversation_commands.steering import (
+    BtwSteering,
+    get_btw_steering,
+)
+
+
+__all__ = [
+    "HELP_TEXT",
+    "BtwAction",
+    "BtwKind",
+    "BtwSteering",
+    "ParsedBtwLine",
+    "btw_commands",
+    "enable_btw_commands",
+    "get_btw_steering",
+    "parse_btw_line",
+]
--- a/lib/crewai/src/crewai/experimental/conversation_commands/addon.py
+++ b/lib/crewai/src/crewai/experimental/conversation_commands/addon.py
@@ -0,0 +1,192 @@
+"""Opt-in installer for experimental ``/btw`` conversational commands."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterator
+from functools import wraps
+from typing import Any, TypeVar, cast, overload
+
+from crewai.experimental.conversation_commands.parser import parse_btw_line
+from crewai.experimental.conversation_commands.steering import (
+    ENABLED_ATTR,
+    apply_btw_action,
+    get_btw_steering,
+)
+from crewai.types.streaming import StreamFrame, StreamSession
+
+
+T = TypeVar("T")
+
+
+class _Intercept:
+    __slots__ = ("consumed", "reply", "user_message")
+
+    def __init__(
+        self,
+        *,
+        consumed: bool,
+        user_message: str | None,
+        reply: str | None = None,
+    ) -> None:
+        self.consumed = consumed
+        self.user_message = user_message
+        self.reply = reply
+
+
+def _intercept(flow: Any, message: str) -> _Intercept:
+    parsed = parse_btw_line(message)
+    if parsed.action is None:
+        return _Intercept(consumed=False, user_message=parsed.user_message)
+
+    reply = apply_btw_action(flow, parsed.action)
+    if parsed.user_message:
+        return _Intercept(
+            consumed=False,
+            user_message=parsed.user_message,
+            reply=reply,
+        )
+    return _Intercept(consumed=True, user_message=None, reply=reply)
+
+
+def _ack_stream(reply: str) -> StreamSession[str]:
+    def frames() -> Iterator[StreamFrame]:
+        return iter(())
+
+    session: StreamSession[str] = StreamSession(sync_iterator=frames())
+    session._set_result(reply)
+    return session
+
+
+def _already_enabled(target: Any) -> bool:
+    if getattr(target, ENABLED_ATTR, False):
+        return True
+    if not isinstance(target, type) and getattr(type(target), ENABLED_ATTR, False):
+        return True
+    return False
+
+
+def _wrap_handle_turn(original: Callable[..., Any]) -> Callable[..., Any]:
+    @wraps(original)
+    def handle_turn(self: Any, message: str, *args: Any, **kwargs: Any) -> Any:
+        decision = _intercept(self, message)
+        if decision.consumed:
+            return decision.reply
+        return original(self, decision.user_message, *args, **kwargs)
+
+    return handle_turn
+
+
+def _wrap_stream_turn(original: Callable[..., Any]) -> Callable[..., Any]:
+    @wraps(original)
+    def stream_turn(self: Any, message: str, *args: Any, **kwargs: Any) -> Any:
+        decision = _intercept(self, message)
+        if decision.consumed:
+            return _ack_stream(decision.reply or "")
+        return original(self, decision.user_message, *args, **kwargs)
+
+    return stream_turn
+
+
+def _wrap_route_turn(original: Callable[..., Any]) -> Callable[..., Any]:
+    @wraps(original)
+    def route_turn(self: Any, context: dict[str, Any]) -> str | None:
+        forced = get_btw_steering(self).consume_forced_route()
+        if forced:
+            return forced
+        return original(self, context)
+
+    return route_turn
+
+
+def _wrap_build_router_context(original: Callable[..., Any]) -> Callable[..., Any]:
+    @wraps(original)
+    def build_router_context(self: Any) -> dict[str, Any]:
+        return get_btw_steering(self).apply_to_router_context(original(self))
+
+    return build_router_context
+
+
+def _wrap_resolve_system_prompt(original: Callable[..., Any]) -> Callable[..., Any]:
+    @wraps(original)
+    def _resolve_system_prompt(self: Any) -> str | None:
+        return get_btw_steering(self).apply_to_system_prompt(original(self))
+
+    return _resolve_system_prompt
+
+
+def _wrap_class(cls: type[T]) -> type[T]:
+    cls.handle_turn = _wrap_handle_turn(cls.handle_turn)  # type: ignore[attr-defined]
+    cls.stream_turn = _wrap_stream_turn(cls.stream_turn)  # type: ignore[attr-defined]
+    cls.route_turn = _wrap_route_turn(cls.route_turn)  # type: ignore[attr-defined]
+    cls.build_router_context = _wrap_build_router_context(  # type: ignore[attr-defined]
+        cls.build_router_context
+    )
+    cls._resolve_system_prompt = _wrap_resolve_system_prompt(  # type: ignore[attr-defined]
+        cls._resolve_system_prompt
+    )
+    setattr(cls, ENABLED_ATTR, True)
+    return cls
+
+
+def _bind_and_wrap(
+    flow: Any,
+    name: str,
+    wrapper: Callable[[Callable[..., Any]], Callable[..., Any]],
+) -> None:
+    original = getattr(type(flow), name)
+    wrapped = wrapper(original)
+
+    @wraps(original)
+    def bound(*args: Any, **kwargs: Any) -> Any:
+        return wrapped(flow, *args, **kwargs)
+
+    object.__setattr__(flow, name, bound)
+
+
+def _wrap_instance(flow: T) -> T:
+    target = cast(Any, flow)
+    _bind_and_wrap(target, "handle_turn", _wrap_handle_turn)
+    _bind_and_wrap(target, "stream_turn", _wrap_stream_turn)
+    _bind_and_wrap(target, "route_turn", _wrap_route_turn)
+    _bind_and_wrap(target, "build_router_context", _wrap_build_router_context)
+    _bind_and_wrap(target, "_resolve_system_prompt", _wrap_resolve_system_prompt)
+    object.__setattr__(target, ENABLED_ATTR, True)
+    return flow
+
+
+def enable_btw_commands(target: T) -> T:
+    """Install ``/btw`` interjections on a Flow class or instance.
+
+    Opt-in only: conversational flows ignore slash lines until this is
+    applied. Safe to call more than once.
+    """
+    if _already_enabled(target):
+        return target
+    if isinstance(target, type):
+        return _wrap_class(cast(type[T], target))
+    return _wrap_instance(target)
+
+
+@overload
+def btw_commands(flow_cls: type[T]) -> type[T]: ...
+
+
+@overload
+def btw_commands(flow_cls: None = None) -> Callable[[type[T]], type[T]]: ...
+
+
+def btw_commands(
+    flow_cls: type[T] | None = None,
+) -> type[T] | Callable[[type[T]], type[T]]:
+    """Class decorator that enables experimental ``/btw`` commands.
+
+    Use as ``@btw_commands`` or ``@btw_commands()`` above
+    ``@ConversationConfig(...)``.
+    """
+
+    def decorate(cls: type[T]) -> type[T]:
+        return enable_btw_commands(cls)
+
+    if flow_cls is not None:
+        return decorate(flow_cls)
+    return decorate
--- a/lib/crewai/src/crewai/experimental/conversation_commands/parser.py
+++ b/lib/crewai/src/crewai/experimental/conversation_commands/parser.py
@@ -0,0 +1,109 @@
+"""Parse ``/btw`` interjections from a conversational user line.
+
+A leading ``/btw`` is a side-channel command: it steers later turns and
+does not become the user utterance. Appending `` /btw …`` to a normal
+message applies the command and still runs the remaining text as a turn.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+import re
+
+
+_LEADING_BTW = re.compile(r"^/btw(?:\s+|$)", re.IGNORECASE)
+_INLINE_BTW = re.compile(r"\s+/btw(?:\s+|$)", re.IGNORECASE)
+_HELP_LINES = frozenset({"/help", "/?"})
+
+HELP_TEXT = (
+    "Experimental /btw commands (side-channel; they do not become a user turn):\n"
+    "  /btw <note>                    Persist a steering note for later turns\n"
+    "  /btw route <label>             Force the next turn onto that route\n"
+    "  /btw route <label> persist     Keep forcing that route until /btw clear\n"
+    "  /btw stay <label>              Same as route <label> persist\n"
+    "  /btw clear                     Drop notes and forced routes\n"
+    "  /btw show                      Show current steering\n"
+    "  /help                          This list\n"
+    "\n"
+    "You can also append a command to a message:\n"
+    "  What's the weather /btw keep it to one sentence"
+)
+
+
+class BtwKind(str, Enum):
+    """Kind of ``/btw`` action parsed from a user line."""
+
+    NOTE = "note"
+    ROUTE = "route"
+    CLEAR = "clear"
+    SHOW = "show"
+    HELP = "help"
+
+
+@dataclass(frozen=True)
+class BtwAction:
+    """A parsed ``/btw`` (or ``/help``) action."""
+
+    kind: BtwKind
+    argument: str = ""
+    persist_route: bool = False
+
+
+@dataclass(frozen=True)
+class ParsedBtwLine:
+    """Split of a user line into an optional command and remaining utterance.
+
+    ``user_message`` is ``None`` when the line is only a command and should
+    not run a conversational turn.
+    """
+
+    action: BtwAction | None
+    user_message: str | None
+
+
+def parse_btw_line(message: str) -> ParsedBtwLine:
+    """Parse a conversational line into a ``/btw`` action and leftover text."""
+    stripped = message.strip()
+    if not stripped:
+        return ParsedBtwLine(action=None, user_message=stripped)
+
+    if stripped.lower() in _HELP_LINES:
+        return ParsedBtwLine(action=BtwAction(kind=BtwKind.HELP), user_message=None)
+
+    leading = _LEADING_BTW.match(stripped)
+    if leading is not None:
+        return ParsedBtwLine(
+            action=_parse_btw_body(stripped[leading.end() :].strip()),
+            user_message=None,
+        )
+
+    inline = _INLINE_BTW.search(stripped)
+    if inline is not None:
+        user = stripped[: inline.start()].strip()
+        return ParsedBtwLine(
+            action=_parse_btw_body(stripped[inline.end() :].strip()),
+            user_message=user or None,
+        )
+
+    return ParsedBtwLine(action=None, user_message=stripped)
+
+
+def _parse_btw_body(body: str) -> BtwAction:
+    if not body:
+        return BtwAction(kind=BtwKind.SHOW)
+
+    parts = body.split()
+    head = parts[0].lower()
+    if head == "clear":
+        return BtwAction(kind=BtwKind.CLEAR)
+    if head in {"show", "status"}:
+        return BtwAction(kind=BtwKind.SHOW)
+    if head == "help":
+        return BtwAction(kind=BtwKind.HELP)
+    if head == "route" and len(parts) >= 2:
+        persist = len(parts) >= 3 and parts[2].lower() in {"persist", "stay"}
+        return BtwAction(kind=BtwKind.ROUTE, argument=parts[1], persist_route=persist)
+    if head == "stay" and len(parts) >= 2:
+        return BtwAction(kind=BtwKind.ROUTE, argument=parts[1], persist_route=True)
+    return BtwAction(kind=BtwKind.NOTE, argument=body)
--- a/lib/crewai/src/crewai/experimental/conversation_commands/steering.py
+++ b/lib/crewai/src/crewai/experimental/conversation_commands/steering.py
@@ -0,0 +1,159 @@
+"""Session-scoped steering applied by experimental ``/btw`` commands."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from crewai.experimental.conversation_commands.parser import (
+    HELP_TEXT,
+    BtwAction,
+    BtwKind,
+)
+from crewai.flow.conversational import ConversationEvent, ConversationState
+
+
+STEERING_ATTR = "_btw_steering"
+ENABLED_ATTR = "_btw_commands_enabled"
+
+_STEERING_HEADER = (
+    "By-the-way steering (follow these unless they conflict with safety):"
+)
+
+
+@dataclass
+class BtwSteering:
+    """In-memory steering notes and optional forced route for one session."""
+
+    notes: list[str] = field(default_factory=list)
+    forced_route: str | None = None
+    persist_route: bool = False
+
+    def apply_note(self, note: str) -> str:
+        text = note.strip()
+        if not text:
+            return self.show()
+        if text not in self.notes:
+            self.notes.append(text)
+        return f"Noted. I'll keep this in mind: {text}"
+
+    def apply_route(self, route: str, *, persist: bool) -> str:
+        label = route.strip()
+        self.forced_route = label
+        self.persist_route = persist
+        if persist:
+            return f"I'll keep using route {label} until you /btw clear."
+        return f"Next turn will use route {label}."
+
+    def clear(self) -> str:
+        self.notes.clear()
+        self.forced_route = None
+        self.persist_route = False
+        return "Cleared steering notes and forced routes."
+
+    def show(self) -> str:
+        lines = ["Current /btw steering:"]
+        if self.notes:
+            lines.extend(f"- {note}" for note in self.notes)
+        else:
+            lines.append("- (no notes)")
+        if self.forced_route:
+            mode = "persist" if self.persist_route else "next turn"
+            lines.append(f"- forced route: {self.forced_route} ({mode})")
+        else:
+            lines.append("- forced route: (none)")
+        return "\n".join(lines)
+
+    def consume_forced_route(self) -> str | None:
+        """Return the forced route, clearing it when it was one-shot."""
+        route = self.forced_route
+        if route is None:
+            return None
+        if not self.persist_route:
+            self.forced_route = None
+        return route
+
+    def apply_to_system_prompt(self, base: str | None) -> str | None:
+        if not self.notes:
+            return base
+        steering = _STEERING_HEADER + "\n" + "\n".join(f"- {note}" for note in self.notes)
+        if base:
+            return f"{base}\n\n{steering}"
+        return steering
+
+    def apply_to_router_context(self, context: dict[str, Any]) -> dict[str, Any]:
+        if not self.notes and not self.forced_route:
+            return context
+        enriched = dict(context)
+        if self.notes:
+            enriched["steering_notes"] = list(self.notes)
+        if self.forced_route:
+            enriched["forced_route"] = self.forced_route
+        return enriched
+
+
+def get_btw_steering(flow: Any) -> BtwSteering:
+    """Return the per-instance steering store, creating it if needed."""
+    steering = getattr(flow, STEERING_ATTR, None)
+    if isinstance(steering, BtwSteering):
+        return steering
+    steering = BtwSteering()
+    object.__setattr__(flow, STEERING_ATTR, steering)
+    return steering
+
+
+def apply_btw_action(flow: Any, action: BtwAction) -> str:
+    """Mutate steering on ``flow`` and return the acknowledgement text."""
+    steering = get_btw_steering(flow)
+    if action.kind is BtwKind.HELP:
+        reply = HELP_TEXT
+    elif action.kind is BtwKind.CLEAR:
+        reply = steering.clear()
+    elif action.kind is BtwKind.SHOW:
+        reply = steering.show()
+    elif action.kind is BtwKind.ROUTE:
+        reply = _apply_route(flow, steering, action)
+    else:
+        reply = steering.apply_note(action.argument)
+    _record_btw_event(flow, action, reply)
+    return reply
+
+
+def _apply_route(flow: Any, steering: BtwSteering, action: BtwAction) -> str:
+    label = action.argument.strip()
+    available = _available_routes(flow)
+    if available is not None and label not in available:
+        listed = ", ".join(sorted(available)) or "(none)"
+        return f"Unknown route {label!r}. Available: {listed}"
+    return steering.apply_route(label, persist=action.persist_route)
+
+
+def _available_routes(flow: Any) -> set[str] | None:
+    resolver = getattr(flow, "_effective_routes", None)
+    if not callable(resolver):
+        return None
+    try:
+        routes = resolver()
+    except Exception:
+        return None
+    if isinstance(routes, set):
+        return routes
+    return set(routes)
+
+
+def _record_btw_event(flow: Any, action: BtwAction, reply: str) -> None:
+    state = getattr(flow, "_state", None) or getattr(flow, "state", None)
+    if not isinstance(state, ConversationState):
+        return
+    state.events.append(
+        ConversationEvent(
+            type="btw_command",
+            payload={
+                "kind": action.kind.value,
+                "argument": action.argument,
+                "persist_route": action.persist_route,
+                "reply": reply,
+            },
+            visibility="private",
+        )
+    )
--- a/lib/crewai/tests/experimental/conversation_commands/test_btw_commands.py
+++ b/lib/crewai/tests/experimental/conversation_commands/test_btw_commands.py
@@ -0,0 +1,209 @@
+"""Experimental /btw commands steer conversational flows without a user turn."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from crewai.experimental.conversation_commands import (
+    HELP_TEXT,
+    BtwKind,
+    btw_commands,
+    enable_btw_commands,
+    get_btw_steering,
+    parse_btw_line,
+)
+from crewai.flow import ConversationConfig, ConversationState, Flow, listen
+
+
+class ConversationalFlow(Flow[ConversationState]):
+    conversational = True
+
+
+class TestParseBtwLine:
+    def test_plain_message_is_unchanged(self) -> None:
+        parsed = parse_btw_line("Where is my order?")
+        assert parsed.action is None
+        assert parsed.user_message == "Where is my order?"
+
+    def test_leading_note_consumes_the_line(self) -> None:
+        parsed = parse_btw_line("/btw keep answers under 20 words")
+        assert parsed.action is not None
+        assert parsed.action.kind is BtwKind.NOTE
+        assert parsed.action.argument == "keep answers under 20 words"
+        assert parsed.user_message is None
+
+    def test_inline_note_keeps_the_utterance(self) -> None:
+        parsed = parse_btw_line("What's the weather /btw keep it to one sentence")
+        assert parsed.action is not None
+        assert parsed.action.kind is BtwKind.NOTE
+        assert parsed.action.argument == "keep it to one sentence"
+        assert parsed.user_message == "What's the weather"
+
+    def test_route_and_persist_forms(self) -> None:
+        once = parse_btw_line("/btw route RESEARCH")
+        assert once.action is not None
+        assert once.action.kind is BtwKind.ROUTE
+        assert once.action.argument == "RESEARCH"
+        assert once.action.persist_route is False
+
+        persist = parse_btw_line("/BTW stay RESEARCH")
+        assert persist.action is not None
+        assert persist.action.persist_route is True
+
+    def test_help_and_bare_btw(self) -> None:
+        help_line = parse_btw_line("/help")
+        assert help_line.action is not None
+        assert help_line.action.kind is BtwKind.HELP
+        assert help_line.user_message is None
+
+        show = parse_btw_line("/btw")
+        assert show.action is not None
+        assert show.action.kind is BtwKind.SHOW
+
+
+@ConversationConfig(defer_trace_finalization=False)
+class _RoutedChat(ConversationalFlow):
+    turns: int = 0
+
+    def route_turn(self, context: dict[str, Any]) -> str | None:
+        message = (self.state.current_user_message or "").lower()
+        if "research" in message:
+            return "RESEARCH"
+        return "work"
+
+    @listen("work")
+    def handle_work(self) -> str:
+        self.turns += 1
+        reply = f"worked: {self.state.current_user_message}"
+        self.append_assistant_message(reply)
+        return reply
+
+    @listen("RESEARCH")
+    def handle_research(self) -> str:
+        self.turns += 1
+        reply = f"researched: {self.state.current_user_message}"
+        self.append_assistant_message(reply)
+        return reply
+
+
+@btw_commands
+@ConversationConfig(defer_trace_finalization=False)
+class _BtwChat(_RoutedChat):
+    pass
+
+
+class TestBtwCommandsOnFlow:
+    def test_note_does_not_run_a_turn_or_append_user_history(self) -> None:
+        flow = _BtwChat()
+        reply = flow.handle_turn("/btw keep answers under 20 words")
+
+        assert "Noted" in reply
+        assert flow.turns == 0
+        assert flow.state.messages == []
+        assert flow.state.current_user_message is None
+        assert get_btw_steering(flow).notes == ["keep answers under 20 words"]
+        assert any(event.type == "btw_command" for event in flow.state.events)
+
+    def test_note_is_injected_into_later_system_prompt_and_router_context(
+        self,
+    ) -> None:
+        flow = _BtwChat()
+        flow.handle_turn("/btw be terse")
+
+        prompt = flow._resolve_system_prompt()
+        assert prompt is not None
+        assert "be terse" in prompt
+
+        context = flow.build_router_context()
+        assert context["steering_notes"] == ["be terse"]
+
+    def test_later_user_turn_still_runs(self) -> None:
+        flow = _BtwChat()
+        flow.handle_turn("/btw be terse")
+        result = flow.handle_turn("hello there")
+
+        assert result == "worked: hello there"
+        assert flow.turns == 1
+        assert flow.state.messages[0].role == "user"
+        assert flow.state.messages[0].content == "hello there"
+
+    def test_forced_route_wins_for_the_next_turn_only(self) -> None:
+        flow = _BtwChat()
+        ack = flow.handle_turn("/btw route RESEARCH")
+        assert "RESEARCH" in ack
+
+        first = flow.handle_turn("hello there")
+        assert first == "researched: hello there"
+        assert flow.state.last_intent == "RESEARCH"
+
+        second = flow.handle_turn("hello again")
+        assert second == "worked: hello again"
+
+    def test_persist_route_keeps_forcing_until_cleared(self) -> None:
+        flow = _BtwChat()
+        flow.handle_turn("/btw stay RESEARCH")
+        assert flow.handle_turn("hello") == "researched: hello"
+        assert flow.handle_turn("again") == "researched: again"
+
+        flow.handle_turn("/btw clear")
+        assert flow.handle_turn("hello") == "worked: hello"
+
+    def test_unknown_route_is_rejected(self) -> None:
+        flow = _BtwChat()
+        reply = flow.handle_turn("/btw route NOPE")
+        assert "Unknown route" in reply
+        assert get_btw_steering(flow).forced_route is None
+        assert flow.turns == 0
+
+    def test_inline_command_steers_the_same_turn(self) -> None:
+        flow = _BtwChat()
+        result = flow.handle_turn("hello there /btw route RESEARCH")
+
+        assert result == "researched: hello there"
+        assert flow.state.messages[0].content == "hello there"
+        assert not any(
+            "/btw" in str(message.content) for message in flow.state.messages
+        )
+
+    def test_help_returns_the_catalog(self) -> None:
+        flow = _BtwChat()
+        reply = flow.handle_turn("/help")
+        assert reply == HELP_TEXT
+        assert flow.turns == 0
+
+    def test_chat_repl_surfaces_standalone_acks(self) -> None:
+        flow = _BtwChat()
+        inputs = iter(["/btw be terse", "hello", "quit"])
+        outputs: list[str] = []
+
+        flow.chat(
+            input_fn=lambda _: next(inputs),
+            output_fn=outputs.append,
+            defer_trace_finalization=False,
+        )
+
+        assert any("Noted" in line for line in outputs)
+        assert any("worked: hello" in line for line in outputs)
+        assert flow.turns == 1
+
+    def test_stream_turn_returns_an_ack_session_for_standalone_commands(self) -> None:
+        flow = _BtwChat()
+        stream = flow.stream_turn("/btw show")
+        with stream:
+            frames = list(stream.events)
+        assert frames == []
+        assert "Current /btw steering" in stream.result
+
+    def test_enable_on_instance_does_not_change_undecorated_classes(self) -> None:
+        plain = _RoutedChat()
+        enabled = enable_btw_commands(_RoutedChat())
+
+        plain_result = plain.handle_turn("/btw be terse")
+        assert plain.turns == 1
+        assert plain.state.messages[0].content == "/btw be terse"
+        assert "worked: /btw be terse" in str(plain_result)
+
+        ack = enabled.handle_turn("/btw be terse")
+        assert "Noted" in ack
+        assert enabled.turns == 0
+        assert enabled.state.messages == []