feat(flow): let a declaration name the router's response format (#7063)

* feat(flow): let a declaration name the router's response format

`conversational.router.response_format` was typed `Any` and dropped with a
warning, because `_router_response_format` hands its value straight to
`llm.call(response_format=...)`, which needs a real class. So the router always
used its synthesized fallback: `intent: str` with the route labels only in a
field description.

The field now takes the same `{"python": "module.path.Class"}` shape a crew
agent's `response_format` uses, resolved through the same
`_resolve_model_class`. That brings the project-root containment with it -- a
declaration cannot reach outside the project to import code -- and gives the
router a `Literal[...]` of the real route labels instead of a bare string.

The DSL projection now emits that shape too, so a live class on a Python flow
round-trips as `{"python": ...}` rather than an opaque `{"ref": ...}` that
nothing could reload.

A bare `module:qualname` ref is now a load-time validation error instead of
being silently discarded; the test that pinned the old drop-with-warning
behavior is updated to assert that.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(flow): do not project a response format that cannot be reloaded

`_python_reference` emitted a dotted path for any class, including two that
cannot be imported back: a non-Pydantic class, and one defined inside a
function, whose `__qualname__` carries `<locals>`. The definition then held a
ref that only failed when something tried to resolve it.

Both are dropped at projection time with a warning naming why, so a reload
falls back to the synthesized response format. The live class still drives the
running flow; only the projection omits it.

Found by CodeRabbit on #7063.

* test(flows): assert the response_format omission warnings

The projection tests only checked for None, so removing the warning that tells
an author their response_format was dropped would still pass.

Found by CodeRabbit on #7063.

* fix(flows): only project a response_format ref that imports back

The check rejected <locals> classes but still emitted a path for a nested one.
The loader splits a ref on its last dot, so module.Outer.Route resolves
module.Outer as a module that does not exist - proven: reload raised
JSONProjectError. A create_model() class held only in a local is unreachable
the same way.

Projection now confirms module.qualname resolves back to the class, against the
already-imported module so it never triggers an import.

Found by CodeRabbit on #7063.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/edge/ar/guides/flows/conversational-flows.mdx

@@ -547,7 +547,7 @@ finally:
 | غير قابل للتعبير | استخدم بدلًا منه |
 |-----------------|-------------|
 | مثيل `LLM` حي أو `BaseLLM` مخصص | سلسلة معرّف نموذج أو خريطة إعدادات ثابتة |
-| `router.response_format` كصنف نموذج | احذفه؛ يولّد الإطار واحدًا. يُتجاهل المرجع أو المخطط مع تحذير |
+| `router.response_format` كصنف نموذج حيّ | سمِّ الصنف بمرجع python: `response_format: {python: my_project.schemas.ConversationRoute}`. احذفه ويولّد الإطار واحدًا |
 | تجاوز `route_turn()` | اكتب Flow بلغة Python، أو استبدل طريقة `route_conversation` التعريفية بإجراء `call: code` / expression |
 | تجاوز `can_answer_from_history()` | اكتب Flow بلغة Python؛ واضبط توجيه السجل القياسي عبر `conversational.answer_from_history_llm` |
 

docs/edge/en/guides/flows/conversational-flows.mdx

@@ -546,7 +546,7 @@ Route labels and method names share one trigger namespace, so a handler must not
 | Not expressible | Use instead |
 |-----------------|-------------|
 | A live `LLM` instance or a custom `BaseLLM` | A model id string or static configuration mapping |
-| `router.response_format` as a model class | Omit it; the framework synthesizes one. A ref or schema is ignored with a warning |
+| `router.response_format` as a live model class | Name the class with a python ref: `response_format: {python: my_project.schemas.ConversationRoute}`. Omit it and the framework synthesizes one |
 | A `route_turn()` override | Author the Flow in Python, or replace the declarative `route_conversation` method with a `call: code` / expression action |
 | A `can_answer_from_history()` override | Author the Flow in Python; configure standard history routing with `conversational.answer_from_history_llm` |
 

docs/edge/ko/guides/flows/conversational-flows.mdx

@@ -543,7 +543,7 @@ finally:
 | 표현 불가 | 대신 사용 |
 |-----------------|-------------|
 | 살아 있는 `LLM` 인스턴스나 커스텀 `BaseLLM` | 모델 id 문자열 또는 정적 설정 mapping |
-| 모델 클래스로서의 `router.response_format` | 생략하세요; 프레임워크가 생성합니다. ref나 스키마는 경고와 함께 무시됩니다 |
+| 살아 있는 모델 클래스로서의 `router.response_format` | python ref로 클래스를 지정하세요: `response_format: {python: my_project.schemas.ConversationRoute}`. 생략하면 프레임워크가 생성합니다 |
 | `route_turn()` 재정의 | Flow를 Python으로 작성하거나 선언적 `route_conversation` 메서드를 `call: code` / expression action으로 교체 |
 | `can_answer_from_history()` 재정의 | Flow를 Python으로 작성하고 표준 기록 라우팅은 `conversational.answer_from_history_llm`으로 설정 |
 

docs/edge/pt-BR/guides/flows/conversational-flows.mdx

@@ -548,7 +548,7 @@ Rótulos de rota e nomes de métodos compartilham um único namespace de gatilho
 | Não expressável | Use no lugar |
 |-----------------|-------------|
 | Uma instância `LLM` viva ou um `BaseLLM` customizado | Um id de modelo em string ou mapping estático de configuração |
-| `router.response_format` como classe de modelo | Omita; o framework sintetiza uma. Um ref ou schema é ignorado com um aviso |
+| `router.response_format` como classe de modelo viva | Nomeie a classe com um ref python: `response_format: {python: my_project.schemas.ConversationRoute}`. Omita e o framework sintetiza uma |
 | Uma sobrescrita de `route_turn()` | Escreva o Flow em Python ou substitua o método declarativo `route_conversation` por uma ação `call: code` / expressão |
 | Uma sobrescrita de `can_answer_from_history()` | Escreva o Flow em Python; configure o roteamento padrão por histórico com `conversational.answer_from_history_llm` |
 

lib/crewai/src/crewai/experimental/conversational_mixin.py

@@ -23,6 +23,7 @@ from contextlib import contextmanager
 from enum import Enum
 import json
 import logging
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast
 
 from pydantic import BaseModel, ConfigDict, Field, create_model
@@ -58,6 +59,7 @@ from crewai.flow.conversational_definition import (
 from crewai.flow.dsl import listen, start
 from crewai.flow.dsl._utils import _method_action, _set_flow_method_definition
 from crewai.flow.flow_definition import FlowDefinition, FlowMethodDefinition
+from crewai.project.crew_definition import PythonReferenceDefinition
 from crewai.utilities.types import LLMMessage
 
 
@@ -83,25 +85,39 @@ def _iter_condition_labels(condition: Any) -> set[str]:
     return set()
 
 
+def _resolve_router_response_format(
+    declared: PythonReferenceDefinition | None,
+    project_root: Path | None,
+) -> type[BaseModel] | None:
+    """Import the model class a declaration names for the routing decision.
+
+    ``_router_response_format`` hands its result straight to
+    ``llm.call(response_format=...)``, which needs a real class. Resolved the
+    same way a crew agent's ``response_format`` is, so the declaration shape
+    and its failure messages match -- including resolving the ref against the
+    declaration's own directory, so a flow loaded by path finds the model
+    sitting next to it rather than under whatever the cwd happens to be.
+    """
+    if declared is None:
+        return None
+
+    from crewai.project.json_loader import _resolve_model_class
+
+    return _resolve_model_class(
+        declared.model_dump(),
+        "conversational.router.response_format",
+        project_root,
+    )
+
+
 def _router_config_from_definition(
     definition: FlowConversationalRouterDefinition,
+    project_root: Path | None,
 ) -> RouterConfig:
     """Build a live ``RouterConfig`` from its serializable form."""
-    response_format = definition.response_format
-    if response_format is not None and not (
-        isinstance(response_format, type) and issubclass(response_format, BaseModel)
-    ):
-        # A declaration can only carry a ``module:qualname`` ref or a schema
-        # dict here, and ``_router_response_format`` hands its result straight
-        # to ``llm.call(response_format=...)``, which needs a real class.
-        # Dropping it falls back to the synthesized single-field model.
-        logger.warning(
-            "Ignoring conversational router response_format %r: a declaration "
-            "cannot carry a model class. The router will use its synthesized "
-            "response format instead.",
-            response_format,
-        )
-        response_format = None
+    response_format = _resolve_router_response_format(
+        definition.response_format, project_root
+    )
 
     return RouterConfig(
         prompt=definition.prompt,
@@ -117,17 +133,19 @@ def _router_config_from_definition(
 
 def _config_from_definition(
     definition: FlowConversationalDefinition,
+    project_root: Path | None,
 ) -> ConversationConfig:
     """Build a live ``ConversationConfig`` from its serializable form.
 
     Used for flows built from a declaration, which have no class-level
-    ``conversational_config`` to read.
+    ``conversational_config`` to read. ``project_root`` is the declaration
+    file's directory, used to resolve the Python refs it names.
     """
     return ConversationConfig(
         system_prompt=definition.system_prompt,
         llm=definition.llm,
         router=(
-            _router_config_from_definition(definition.router)
+            _router_config_from_definition(definition.router, project_root)
             if definition.router is not None
             else None
         ),
@@ -941,11 +959,12 @@ class _ConversationalMixin:
         if resolved is not None:
             return resolved
 
-        definition = self._conversation_definition
+        flow_definition = self._conversation_flow_definition()
+        definition = flow_definition.conversational
         if definition is None or not definition.enabled:
             return None
 
-        resolved = _config_from_definition(definition)
+        resolved = _config_from_definition(definition, flow_definition.source_dir)
         object.__setattr__(self, "_resolved_conversation_config", resolved)
         return resolved
 

lib/crewai/src/crewai/flow/conversational_definition.py

@@ -11,12 +11,22 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, Field
 
+from crewai.project.crew_definition import PythonReferenceDefinition
+
 
 class FlowConversationalRouterDefinition(BaseModel):
     """Static conversational router configuration."""
 
     prompt: str | None = None
-    response_format: Any = None
+    response_format: PythonReferenceDefinition | None = Field(
+        default=None,
+        description=(
+            "Optional Python reference to a Pydantic model for the routing "
+            "decision. Omit it and the framework synthesizes one from the "
+            "route labels."
+        ),
+        examples=[{"python": "my_project.schemas.ConversationRoute"}],
+    )
     llm: Any = Field(
         default=None,
         description=(

lib/crewai/src/crewai/flow/dsl/_utils.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import logging
+import sys
 from typing import Any, ParamSpec, TypeVar
 
 from pydantic import BaseModel, ValidationError
@@ -276,6 +277,50 @@ def _build_persistence_definition(value: Any) -> FlowPersistenceDefinition | Non
     )
 
 
+def _resolves_to_itself(cls: type[BaseModel]) -> bool:
+    """Whether ``module.qualname`` would actually import ``cls`` back.
+
+    The loader splits a ref on its last dot, so a nested class projects
+    ``module.Outer.Route`` and ``module.Outer`` is then looked up as a module
+    that does not exist. A ``create_model()`` class held only in a local or a
+    dict is not reachable under its qualname either. Checked against the
+    already-imported module so projecting never triggers an import.
+    """
+    module = sys.modules.get(cls.__module__)
+    return module is not None and getattr(module, cls.__qualname__, None) is cls
+
+
+def _python_reference(value: Any) -> dict[str, str] | None:
+    """Project a class as the dotted ``{"python": ...}`` ref the contract uses.
+
+    The same shape a crew agent's ``response_format`` is declared with, so a
+    projected definition reloads into a live class instead of an opaque ref.
+    """
+    if value is None:
+        return None
+
+    reason: str | None = None
+    if not isinstance(value, type) or not issubclass(value, BaseModel):
+        reason = "is not a Pydantic model class"
+    elif "<locals>" in value.__qualname__:
+        # A dotted path through ``<locals>`` cannot be imported back, so the
+        # ref would only fail when something tried to reload it.
+        reason = "is defined inside a function and cannot be imported by path"
+    elif not _resolves_to_itself(value):
+        reason = "cannot be imported by its module path"
+
+    if reason is not None:
+        logger.warning(
+            "Conversational router response_format %r %s; dropping it from the "
+            "definition. The router will use its synthesized response format.",
+            value,
+            reason,
+        )
+        return None
+
+    return {"python": f"{value.__module__}.{value.__qualname__}"}
+
+
 def _build_conversational_router_definition(
     router_config: Any,
     path: str,
@@ -286,9 +331,8 @@ def _build_conversational_router_definition(
     routes = getattr(router_config, "routes", None)
     return FlowConversationalRouterDefinition(
         prompt=getattr(router_config, "prompt", None),
-        response_format=_serialize_static_value(
-            getattr(router_config, "response_format", None),
-            f"{path}.response_format",
+        response_format=_python_reference(
+            getattr(router_config, "response_format", None)
         ),
         llm=_serialize_static_value(getattr(router_config, "llm", None), f"{path}.llm"),
         routes=[str(route) for route in routes] if routes is not None else None,

lib/crewai/tests/test_flow_conversation.py

@@ -3,12 +3,16 @@
 from __future__ import annotations
 
 import logging
+import sys
 from typing import Any, ClassVar, Literal
 from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 import pytest
-from pydantic import BaseModel
+from pathlib import Path
+import yaml
+
+from pydantic import BaseModel, ValidationError, create_model
 
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.listeners.tracing.trace_listener import TraceCollectionListener
@@ -2303,6 +2307,13 @@ class _ScriptedLLM(BaseLLM):
         return 8192
 
 
+class _Outer:
+    """Holds a nested model, so its qualname is dotted without `<locals>`."""
+
+    class Route(BaseModel):
+        intent: str
+
+
 def _conversational_declaration(**overrides: Any) -> dict[str, Any]:
     """A declaration naming the built-in methods explicitly.
 
@@ -2404,20 +2415,236 @@ class TestDeclarativeConversationalFlow:
         assert ordered == ["bootstrap", "route_conversation"]
         assert sequential is True
 
-    def test_router_response_format_ref_is_dropped_with_a_warning(self, caplog) -> None:
-        caplog.set_level(
-            logging.WARNING, logger="crewai.experimental.conversational_mixin"
-        )
+    def test_router_response_format_takes_a_python_ref_not_a_module_ref(self) -> None:
+        """The contract declares the same `{"python": ...}` shape crews use.
+
+        A bare `module:qualname` ref used to be accepted and silently dropped;
+        it is now a load-time error, so a typo cannot look like it worked.
+        """
+        with pytest.raises(ValidationError, match="response_format"):
+            Flow.from_declaration(
+                contents=_conversational_declaration(
+                    conversational={
+                        "router": {"response_format": {"ref": "some.module:Schema"}}
+                    }
+                )
+            )
+
+
+class TestDeclaredRouterResponseFormat:
+    """A declaration can name the model the routing decision is parsed into."""
+
+    ROUTE_MODULE = (
+        "from typing import Literal\n"
+        "from pydantic import BaseModel\n"
+        "\n"
+        "class ConversationRoute(BaseModel):\n"
+        "    intent: Literal['order', 'converse']\n"
+    )
+
+    @staticmethod
+    def _declaration(router: dict[str, Any]) -> dict[str, Any]:
+        return _conversational_declaration(conversational={"router": router})
+
+    def test_a_declared_python_ref_is_resolved_to_the_class(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        (tmp_path / "routes.py").write_text(self.ROUTE_MODULE, encoding="utf-8")
+        monkeypatch.chdir(tmp_path)
+        monkeypatch.syspath_prepend(str(tmp_path))
+
         flow = Flow.from_declaration(
-            contents=_conversational_declaration(
-                conversational={
-                    "router": {"response_format": {"ref": "some.module:Schema"}}
-                }
+            contents=self._declaration(
+                {"response_format": {"python": "routes.ConversationRoute"}}
+            )
+        )
+        resolved = flow._conversation_config.router.response_format
+
+        assert resolved is not None
+        assert resolved.__name__ == "ConversationRoute"
+        assert "intent" in resolved.model_fields
+        assert flow._router_response_format(flow._conversation_config.router) is resolved
+
+    def test_a_ref_resolves_next_to_the_declaration_not_the_cwd(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A flow loaded by path finds the model sitting beside its YAML.
+
+        Resolution used to fall back to ``Path.cwd()``, so the same project
+        loaded from another directory could not import its own route model.
+        """
+        project = tmp_path / "project"
+        project.mkdir()
+        (project / "chat_routes.py").write_text(self.ROUTE_MODULE, encoding="utf-8")
+        declaration = project / "chat.yaml"
+        declaration.write_text(
+            yaml.safe_dump(
+                self._declaration(
+                    {"response_format": {"python": "chat_routes.ConversationRoute"}}
+                )
+            ),
+            encoding="utf-8",
+        )
+
+        elsewhere = tmp_path / "elsewhere"
+        elsewhere.mkdir()
+        monkeypatch.chdir(elsewhere)
+        monkeypatch.delitem(sys.modules, "chat_routes", raising=False)
+
+        try:
+            flow = Flow.from_declaration(path=declaration)
+            resolved = flow._conversation_config.router.response_format
+        finally:
+            sys.modules.pop("chat_routes", None)
+
+        assert resolved is not None
+        assert resolved.__name__ == "ConversationRoute"
+        assert "intent" in resolved.model_fields
+
+    def test_omitting_it_still_synthesizes_one(self) -> None:
+        flow = Flow.from_declaration(contents=self._declaration({}))
+
+        synthesized = flow._router_response_format(flow._conversation_config.router)
+
+        assert flow._conversation_config.router.response_format is None
+        assert list(synthesized.model_fields) == ["intent"]
+
+    def test_a_ref_outside_the_project_root_is_refused(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Declarations may not reach outside the project to import code."""
+        monkeypatch.chdir(tmp_path)
+        flow = Flow.from_declaration(
+            contents=self._declaration(
+                {"response_format": {"python": "os.path.basename"}}
             )
         )
 
-        assert flow._conversation_config.router.response_format is None
-        assert "cannot carry a model class" in caplog.text
+        # Resolution is lazy: the declaration loads, the import is refused.
+        with pytest.raises(Exception, match="inside the project root"):
+            _ = flow._conversation_config
+
+    def test_a_ref_without_a_dot_is_rejected_at_load(self) -> None:
+        with pytest.raises(ValidationError):
+            Flow.from_declaration(
+                contents=self._declaration({"response_format": {"python": "nodots"}})
+            )
+
+    def test_a_live_class_on_a_python_flow_is_untouched(self) -> None:
+        class MyRoute(BaseModel):
+            intent: str
+
+        @ConversationConfig(router=RouterConfig(response_format=MyRoute))
+        class ClassChat(Flow[ConversationState]):
+            pass
+
+        assert ClassChat()._conversation_config.router.response_format is MyRoute
+
+    def test_a_function_local_model_is_omitted_from_the_projection(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A path through `<locals>` cannot be imported back, so never emit it."""
+
+        class LocalRoute(BaseModel):
+            intent: str
+
+        @ConversationConfig(router=RouterConfig(response_format=LocalRoute))
+        class ClassChat(Flow[ConversationState]):
+            pass
+
+        with caplog.at_level(logging.WARNING, logger="crewai.flow.dsl._utils"):
+            definition = ClassChat.flow_definition()
+
+        assert definition.conversational.router.response_format is None
+        # Silently dropping it would leave the author guessing, so warn.
+        assert "cannot be imported by path" in caplog.text
+        # The live class still drives the running flow; only the projection drops it.
+        assert ClassChat()._conversation_config.router.response_format is LocalRoute
+
+    def test_a_non_model_response_format_is_omitted_from_the_projection(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        class NotAModel:
+            pass
+
+        @ConversationConfig(router=RouterConfig(response_format=NotAModel))
+        class ClassChat(Flow[ConversationState]):
+            pass
+
+        with caplog.at_level(logging.WARNING, logger="crewai.flow.dsl._utils"):
+            definition = ClassChat.flow_definition()
+
+        assert definition.conversational.router.response_format is None
+        assert "is not a Pydantic model class" in caplog.text
+
+    def test_a_nested_model_is_omitted_from_the_projection(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """`module.Outer.Route` reloads `module.Outer` as a module that is absent."""
+
+        @ConversationConfig(router=RouterConfig(response_format=_Outer.Route))
+        class ClassChat(Flow[ConversationState]):
+            pass
+
+        with caplog.at_level(logging.WARNING, logger="crewai.flow.dsl._utils"):
+            definition = ClassChat.flow_definition()
+
+        assert definition.conversational.router.response_format is None
+        assert "cannot be imported by its module path" in caplog.text
+        assert ClassChat()._conversation_config.router.response_format is _Outer.Route
+
+    def test_an_unbound_generated_model_is_omitted_from_the_projection(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A `create_model()` class no module attribute names cannot be reloaded."""
+        generated = create_model("GeneratedRoute", intent=(str, ...))
+
+        @ConversationConfig(router=RouterConfig(response_format=generated))
+        class ClassChat(Flow[ConversationState]):
+            pass
+
+        with caplog.at_level(logging.WARNING, logger="crewai.flow.dsl._utils"):
+            definition = ClassChat.flow_definition()
+
+        assert definition.conversational.router.response_format is None
+        assert "cannot be imported by its module path" in caplog.text
+
+        reloaded = Flow.from_declaration(contents=definition.to_dict())
+        synthesized = reloaded._router_response_format(
+            reloaded._conversation_config.router
+        )
+
+        assert synthesized is not generated
+        assert list(synthesized.model_fields) == ["intent"]
+
+    def test_a_dropped_projection_reloads_with_the_synthesized_model(self) -> None:
+        class LocalRoute(BaseModel):
+            intent: str
+
+        @ConversationConfig(router=RouterConfig(response_format=LocalRoute))
+        class ClassChat(Flow[ConversationState]):
+            pass
+
+        reloaded = Flow.from_declaration(
+            contents=ClassChat.flow_definition().to_dict()
+        )
+        synthesized = reloaded._router_response_format(
+            reloaded._conversation_config.router
+        )
+
+        assert list(synthesized.model_fields) == ["intent"]
+
+    def test_a_live_class_projects_as_a_python_ref(self) -> None:
+        @ConversationConfig(router=RouterConfig(response_format=ConversationState))
+        class ClassChat(Flow[ConversationState]):
+            pass
+
+        projected = ClassChat.flow_definition().conversational.router.response_format
+
+        assert projected is not None
+        assert projected.python == (
+            "crewai.experimental.conversational.ConversationState"
+        )
 
 
 class DeclaredSchemaChatState(ConversationState):