mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-09-22 02:46:39 +00:00
f68fd9e850
* feat(flow): let a chat flow declare its own state shape
A conversational declaration could only use `state: {type: pydantic, ref: ...}`
pointing at a `ConversationState` subclass. Every other shape loaded clean and
then died on the first turn -- inline `json_schema` and a non-subclass ref with
`AttributeError: 'StateWithId' object has no attribute 'messages'`, and
`type: dict` with `AttributeError: 'dict' object has no attribute 'id'`.
`Flow._compose_extension_state_model` is a new runtime extension seam -- the
seventh alongside the existing six -- applied to the model built from `state:`
before the engine wraps it for `id`. The conversational mixin uses it to add
the chat fields to whatever the declaration asked for, so declared fields and
defaults survive; a model that already extends `ConversationState` is returned
untouched, so today's supported shape is a no-op.
`dict` and `unknown` state cannot carry those fields at all, so the default
extension state supplies the real shape (seeded from the declared defaults
where they fit) rather than forbidding it. Raising instead would break
construction, and `Flow[dict]` with `conversational = True` constructs today --
`crewai flow plot` and definition-only consumers would stop working.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
* fix(flow): keep declared defaults and cover an unbuildable state model
Two review follow-ups on the declared-state work.
A `dict` state's defaults are arbitrary keys, and the fallback kept only the
ones matching `ConversationState`, so `{"type": "dict", "default": {"topic":
"ai"}}` lost `topic` before the first turn and an action reading `state.topic`
would fail. They are carried as extras now.
And a declared `pydantic`/`json_schema` state whose model cannot be built --
a bad ref, an invalid schema -- fell through to a plain dict with none of the
chat fields, so the turn died on `state.id` instead. The engine now re-asks the
extension in that case, as if nothing had been declared.
Found by Cursor and CodeRabbit on #7061.
* refactor(flows): drop the unreachable extension-state fallback
The _initial_state_t branch sat after an unconditional return. The
state_definition is None case and _conversation_state_with_defaults now cover
every path that used to reach it.
Found by Cursor on #7061.
---------
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
3418 lines
123 KiB
Python
3418 lines
123 KiB
Python
"""Tests for conversational Flow helpers and kickoff parameters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, ClassVar, Literal
|
|
from unittest.mock import MagicMock, patch
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
from crewai.events.event_bus import crewai_event_bus
|
|
from crewai.events.listeners.tracing.trace_listener import TraceCollectionListener
|
|
from crewai.events.types.flow_events import (
|
|
ConversationMessageAddedEvent,
|
|
ConversationRouteSelectedEvent,
|
|
ConversationTurnCompletedEvent,
|
|
ConversationTurnFailedEvent,
|
|
ConversationTurnStartedEvent,
|
|
FlowStartedEvent,
|
|
MethodExecutionFinishedEvent,
|
|
MethodExecutionStartedEvent,
|
|
)
|
|
from crewai.events.types.llm_events import LLMCallStartedEvent, LLMStreamChunkEvent
|
|
from crewai.experimental import (
|
|
ConversationConfig,
|
|
ConversationMessage,
|
|
ConversationState,
|
|
RouterConfig,
|
|
)
|
|
from crewai.flow import Flow, ChatState, listen, start
|
|
from crewai.flow.async_feedback import HumanFeedbackPending, PendingFeedbackContext
|
|
from crewai.flow.flow_context import (
|
|
current_flow_defer_trace_finalization,
|
|
current_flow_id,
|
|
current_flow_name,
|
|
)
|
|
from crewai.llms.base_llm import BaseLLM
|
|
from crewai.flow.conversation import (
|
|
append_message,
|
|
get_conversation_messages,
|
|
normalize_kickoff_inputs,
|
|
prepare_conversational_turn,
|
|
)
|
|
|
|
class ConversationalFlow(Flow[ConversationState]):
|
|
"""Test base: a ``Flow[ConversationState]`` with conversational mode enabled.
|
|
|
|
Mirrors the documented ``class MyChat(Flow): conversational = True`` pattern
|
|
so the conversational subclasses below stay terse.
|
|
"""
|
|
|
|
conversational = True
|
|
|
|
|
|
class SimpleChatFlow(Flow[ChatState]):
|
|
@start()
|
|
def begin(self):
|
|
return "done"
|
|
|
|
|
|
class DictChatFlow(Flow):
|
|
@start()
|
|
def begin(self):
|
|
return self.state.get("marker", "ok")
|
|
|
|
|
|
class TestNormalizeKickoffInputs:
|
|
def test_merges_session_and_user_message(self) -> None:
|
|
merged = normalize_kickoff_inputs(
|
|
{"foo": 1},
|
|
user_message="hello",
|
|
session_id="sess-1",
|
|
)
|
|
assert merged["id"] == "sess-1"
|
|
assert merged["user_message"] == "hello"
|
|
assert merged["foo"] == 1
|
|
|
|
|
|
class TestMessageHelpers:
|
|
def test_append_message_on_pydantic_state(self) -> None:
|
|
flow = SimpleChatFlow()
|
|
flow._state = ChatState()
|
|
append_message(flow, "user", "hi")
|
|
assert get_conversation_messages(flow) == [{"role": "user", "content": "hi"}]
|
|
|
|
def test_append_message_fallback_buffer(self) -> None:
|
|
flow = DictChatFlow()
|
|
|
|
class _State:
|
|
id = str(uuid4())
|
|
|
|
flow._state = _State()
|
|
append_message(flow, "assistant", "reply")
|
|
assert get_conversation_messages(flow) == [
|
|
{"role": "assistant", "content": "reply"}
|
|
]
|
|
assert flow._conversation_messages == [
|
|
{"role": "assistant", "content": "reply"}
|
|
]
|
|
|
|
|
|
class TestIntentPerTurn:
|
|
def test_prepare_clears_stale_last_intent(self) -> None:
|
|
flow = SimpleChatFlow()
|
|
flow._state = ChatState(last_intent="ORDER", messages=[])
|
|
prepare_conversational_turn(flow, user_message="hello")
|
|
assert flow.state.last_intent is None
|
|
|
|
|
|
class TestClassifyIntent:
|
|
def test_uses_collapse_with_context(self) -> None:
|
|
flow = SimpleChatFlow()
|
|
flow._state = ChatState(
|
|
messages=[{"role": "user", "content": "prior"}],
|
|
)
|
|
|
|
with patch.object(flow, "_collapse_to_outcome", return_value="help") as mock:
|
|
outcome = flow.classify_intent(
|
|
"I need help",
|
|
["order", "help"],
|
|
llm="gpt-4o-mini",
|
|
context=flow.conversation_messages,
|
|
)
|
|
|
|
assert outcome == "help"
|
|
assert "I need help" in mock.call_args[0][0]
|
|
|
|
|
|
class TestConversationalFlow:
|
|
def test_stream_turn_emits_ordered_conversation_frames(self) -> None:
|
|
flow = ConversationalFlow()
|
|
flow.stream = True
|
|
stream_values_seen_by_kickoff: list[bool] = []
|
|
|
|
def kickoff_side_effect(*_: Any, **__: Any) -> str:
|
|
stream_values_seen_by_kickoff.append(flow.stream)
|
|
crewai_event_bus.emit(
|
|
flow,
|
|
LLMStreamChunkEvent(
|
|
type="llm_stream_chunk",
|
|
chunk="pong",
|
|
call_id="call-1",
|
|
),
|
|
)
|
|
return "pong"
|
|
|
|
with patch.object(flow, "kickoff", side_effect=kickoff_side_effect):
|
|
stream = flow.stream_turn("ping", session_id="session-1")
|
|
|
|
with pytest.raises(RuntimeError, match="Streaming has not completed yet"):
|
|
_ = stream.result
|
|
|
|
frames = list(stream.events)
|
|
|
|
assert stream.result == "pong"
|
|
assert stream_values_seen_by_kickoff == [False]
|
|
assert flow.stream is True
|
|
assert [frame.seq for frame in frames] == sorted(frame.seq for frame in frames)
|
|
assert [frame.type for frame in frames] == [
|
|
"conversation_turn_started",
|
|
"llm_stream_chunk",
|
|
"conversation_message_added",
|
|
"conversation_turn_completed",
|
|
]
|
|
assert [frame.channel for frame in frames] == [
|
|
"flow",
|
|
"llm",
|
|
"messages",
|
|
"flow",
|
|
]
|
|
assert frames[1].data["chunk"] == "pong"
|
|
assert flow.state.messages[-1].content == "pong"
|
|
|
|
def test_stream_turn_enables_streaming_on_conversation_llm(self) -> None:
|
|
class FakeLLM(BaseLLM):
|
|
stream_values: ClassVar[list[bool | None]] = []
|
|
|
|
def call(self, messages: Any, *args: Any, **kwargs: Any) -> str:
|
|
self.stream_values.append(self._effective_stream())
|
|
for chunk in ("po", "ng"):
|
|
crewai_event_bus.emit(
|
|
flow,
|
|
LLMStreamChunkEvent(
|
|
type="llm_stream_chunk",
|
|
chunk=chunk,
|
|
call_id="call-1",
|
|
),
|
|
)
|
|
return "pong"
|
|
|
|
FakeLLM.stream_values = []
|
|
llm = FakeLLM(model="gpt-4o-mini", stream=False)
|
|
|
|
@ConversationConfig(llm=llm)
|
|
class StreamingChatFlow(ConversationalFlow):
|
|
pass
|
|
|
|
flow = StreamingChatFlow()
|
|
stream = flow.stream_turn("ping", session_id="session-1")
|
|
frames = list(stream.events)
|
|
|
|
assert stream.result == "pong"
|
|
assert llm.stream_values == [True]
|
|
assert llm.stream is False
|
|
assert [
|
|
frame.data["chunk"]
|
|
for frame in frames
|
|
if frame.type == "llm_stream_chunk"
|
|
] == ["po", "ng"]
|
|
|
|
def test_stream_turn_returns_pending_feedback_without_failure_event(self) -> None:
|
|
flow = ConversationalFlow()
|
|
pending = HumanFeedbackPending(
|
|
context=PendingFeedbackContext(
|
|
flow_id="session-1",
|
|
flow_class="tests.PendingFeedbackFlow",
|
|
method_name="review",
|
|
method_output="draft",
|
|
message="Please review",
|
|
)
|
|
)
|
|
|
|
def kickoff_side_effect(*_: Any, **__: Any) -> None:
|
|
raise pending
|
|
|
|
with patch.object(flow, "kickoff", side_effect=kickoff_side_effect):
|
|
stream = flow.stream_turn("review this", session_id="session-1")
|
|
frames = list(stream.events)
|
|
|
|
assert stream.result is pending
|
|
assert [frame.type for frame in frames] == ["conversation_turn_started"]
|
|
|
|
def test_deferred_multi_turn_emits_single_flow_finished(self) -> None:
|
|
"""A deferred multi-turn session lands as one trace: exactly one
|
|
``FlowFinishedEvent`` is emitted at ``finalize_session_traces()``, not
|
|
one per turn. (Each turn still opens its own ``flow_started``.)
|
|
"""
|
|
from crewai.events.types.flow_events import FlowFinishedEvent
|
|
|
|
@ConversationConfig(defer_trace_finalization=True)
|
|
class TraceFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
reply = f"worked: {self.state.current_user_message}"
|
|
self.append_assistant_message(reply)
|
|
return reply
|
|
|
|
flow = TraceFlow()
|
|
finished: list[FlowFinishedEvent] = []
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def capture(_: Any, event: FlowFinishedEvent) -> None:
|
|
finished.append(event)
|
|
|
|
flow.handle_turn("research apple stock")
|
|
flow.handle_turn("research google stock")
|
|
crewai_event_bus.flush()
|
|
assert finished == [], "deferred turns must not emit per-turn flow_finished"
|
|
|
|
flow.finalize_session_traces()
|
|
crewai_event_bus.flush()
|
|
|
|
assert len(finished) == 1, (
|
|
"a deferred session must emit exactly one flow_finished at finalize"
|
|
)
|
|
|
|
|
|
def test_handle_turn_routes_to_listener_and_records_public_result(self) -> None:
|
|
@ConversationConfig(default_intents=["research"], intent_llm="gpt-4o-mini")
|
|
class ResearchFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_agent_result(
|
|
"researcher",
|
|
"researched answer",
|
|
visibility="public",
|
|
)
|
|
return "researched answer"
|
|
|
|
flow = ResearchFlow()
|
|
|
|
with patch.object(flow, "_collapse_to_outcome", return_value="research"):
|
|
result = flow.handle_turn("research CrewAI")
|
|
|
|
assert result == "researched answer"
|
|
assert flow.state.current_user_message == "research CrewAI"
|
|
assert flow.state.last_intent == "research"
|
|
assert [message.role for message in flow.state.messages] == [
|
|
"user",
|
|
"assistant",
|
|
]
|
|
assert flow.state.messages[-1].content == "researched answer"
|
|
assert flow.state.events[0].agent_name == "researcher"
|
|
assert flow.state.events[0].visibility == "public"
|
|
|
|
def test_private_agent_results_stay_out_of_shared_history(self) -> None:
|
|
class PrivateFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> None:
|
|
self.append_agent_result("planner", "private scratch")
|
|
|
|
flow = PrivateFlow()
|
|
flow.handle_turn("plan quietly")
|
|
|
|
assert [message.role for message in flow.state.messages] == ["user"]
|
|
assert flow.state.events[0].visibility == "private"
|
|
assert flow.state.agent_threads["planner"][0].content == "private scratch"
|
|
|
|
def test_answer_from_history_uses_configured_llm_and_appends_reply(self) -> None:
|
|
@ConversationConfig(answer_from_history_llm="gpt-4o-mini")
|
|
class HistoryFlow(ConversationalFlow):
|
|
pass
|
|
|
|
flow = HistoryFlow()
|
|
flow._state = ConversationState(
|
|
messages=[
|
|
ConversationMessage(role="user", content="research topic"),
|
|
ConversationMessage(role="assistant", content="prior findings"),
|
|
]
|
|
)
|
|
llm = MagicMock()
|
|
llm.call.return_value = "summary from history"
|
|
|
|
with (
|
|
patch.object(
|
|
flow,
|
|
"_collapse_to_outcome",
|
|
return_value="answer_from_history",
|
|
),
|
|
patch.object(flow, "_coerce_llm", return_value=llm),
|
|
):
|
|
result = flow.handle_turn("summarize this")
|
|
|
|
assert result == "summary from history"
|
|
assert flow.state.messages[-1].role == "assistant"
|
|
assert flow.state.messages[-1].content == "summary from history"
|
|
llm.call.assert_called_once()
|
|
|
|
def test_router_config_uses_structured_intent_response(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: Literal["research", "clarify"]
|
|
|
|
llm = MagicMock()
|
|
llm.call.return_value = ResearchRoute(intent="research")
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
prompt="Classify the next action.",
|
|
response_format=ResearchRoute,
|
|
llm=llm,
|
|
routes=["research", "clarify"],
|
|
default_intent="clarify",
|
|
fallback_intent="clarify",
|
|
)
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_assistant_message("researched")
|
|
return "researched"
|
|
|
|
@listen("clarify")
|
|
def ask_clarification(self) -> str:
|
|
self.append_assistant_message("clarify")
|
|
return "clarify"
|
|
|
|
flow = RoutedFlow()
|
|
result = flow.handle_turn("research CrewAI")
|
|
|
|
assert result == "researched"
|
|
llm.call.assert_called_once()
|
|
assert llm.call.call_args.kwargs["response_format"] is ResearchRoute
|
|
assert flow.state.messages[-1].content == "researched"
|
|
|
|
def test_router_config_falls_back_for_invalid_intent(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: str
|
|
|
|
llm = MagicMock()
|
|
llm.call.return_value = ResearchRoute(intent="unknown")
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
prompt="Classify the next action.",
|
|
response_format=ResearchRoute,
|
|
llm=llm,
|
|
routes=["research", "clarify"],
|
|
default_intent="clarify",
|
|
fallback_intent="clarify",
|
|
)
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_assistant_message("researched")
|
|
return "researched"
|
|
|
|
@listen("clarify")
|
|
def ask_clarification(self) -> str:
|
|
self.append_assistant_message("clarify")
|
|
return "clarify"
|
|
|
|
flow = RoutedFlow()
|
|
result = flow.handle_turn("something vague")
|
|
|
|
assert result == "clarify"
|
|
assert flow.state.messages[-1].content == "clarify"
|
|
|
|
def test_router_effective_routes_include_builtins(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: Literal["research", "converse", "end"]
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
prompt="Classify.",
|
|
response_format=ResearchRoute,
|
|
routes=["research"],
|
|
)
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
return "researched"
|
|
|
|
flow = RoutedFlow()
|
|
|
|
assert flow._effective_routes(flow.conversational_config.router) == {
|
|
"research",
|
|
"converse",
|
|
"end",
|
|
}
|
|
|
|
def test_router_infers_custom_routes_without_internal_routes(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: Literal["research", "converse", "end"]
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
prompt="Classify.",
|
|
response_format=ResearchRoute,
|
|
)
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
return "researched"
|
|
|
|
flow = RoutedFlow()
|
|
|
|
assert flow._effective_routes(flow.conversational_config.router) == {
|
|
"research",
|
|
"converse",
|
|
"end",
|
|
}
|
|
|
|
def test_router_config_uses_conversational_defaults(self) -> None:
|
|
llm = MagicMock()
|
|
|
|
@ConversationConfig(
|
|
llm=llm,
|
|
router=RouterConfig(),
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_assistant_message("researched")
|
|
return "researched"
|
|
|
|
flow = RoutedFlow()
|
|
response_format = flow._router_response_format(flow.conversational_config.router)
|
|
llm.call.return_value = response_format(intent="research")
|
|
|
|
result = flow.handle_turn("research CrewAI")
|
|
|
|
assert result == "researched"
|
|
llm.call.assert_called_once()
|
|
assert llm.call.call_args.kwargs["response_format"].__name__ == (
|
|
"ConversationRoute"
|
|
)
|
|
assert flow.state.messages[-1].content == "researched"
|
|
|
|
def test_builtin_converse_appends_assistant_message_and_uses_history(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: Literal["research", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.return_value = ResearchRoute(intent="converse")
|
|
chat_llm = MagicMock()
|
|
chat_llm.call.return_value = "summary from built-in converse"
|
|
|
|
@ConversationConfig(
|
|
system_prompt="You are a helpful research assistant.",
|
|
llm=chat_llm,
|
|
router=RouterConfig(
|
|
prompt="Classify.",
|
|
response_format=ResearchRoute,
|
|
llm=router_llm,
|
|
routes=["research"],
|
|
default_intent="converse",
|
|
),
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_agent_result(
|
|
"researcher",
|
|
"prior findings",
|
|
visibility="public",
|
|
)
|
|
return "prior findings"
|
|
|
|
flow = RoutedFlow()
|
|
flow.state.messages = [
|
|
ConversationMessage(role="user", content="research CrewAI"),
|
|
ConversationMessage(role="assistant", content="prior findings"),
|
|
]
|
|
result = flow.handle_turn("summarize findings")
|
|
|
|
assert result == "summary from built-in converse"
|
|
assert flow.state.messages[-1].content == "summary from built-in converse"
|
|
messages = chat_llm.call.call_args.kwargs["messages"]
|
|
assert messages[0] == {
|
|
"role": "system",
|
|
"content": "You are a helpful research assistant.",
|
|
}
|
|
assert any(message["content"] == "prior findings" for message in messages)
|
|
assert any(message["content"] == "summarize findings" for message in messages)
|
|
|
|
def test_conversational_turn_emits_message_and_route_events(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: Literal["research", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.return_value = ResearchRoute(intent="converse")
|
|
chat_llm = MagicMock()
|
|
chat_llm.call.return_value = "hello back"
|
|
|
|
@ConversationConfig(
|
|
llm=chat_llm,
|
|
router=RouterConfig(
|
|
response_format=ResearchRoute,
|
|
llm=router_llm,
|
|
routes=["research"],
|
|
),
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_assistant_message("researched")
|
|
return "researched"
|
|
|
|
messages: list[ConversationMessageAddedEvent] = []
|
|
routes: list[ConversationRouteSelectedEvent] = []
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(ConversationMessageAddedEvent)
|
|
def capture_message(_: Any, event: ConversationMessageAddedEvent) -> None:
|
|
messages.append(event)
|
|
|
|
@crewai_event_bus.on(ConversationRouteSelectedEvent)
|
|
def capture_route(_: Any, event: ConversationRouteSelectedEvent) -> None:
|
|
routes.append(event)
|
|
|
|
flow = RoutedFlow()
|
|
flow.handle_turn("just chat")
|
|
crewai_event_bus.flush()
|
|
|
|
assert [(event.role, event.content) for event in messages] == [
|
|
("user", "just chat"),
|
|
("assistant", "hello back"),
|
|
]
|
|
assert [event.message_index for event in messages] == [0, 1]
|
|
assert len(routes) == 1
|
|
assert routes[0].route == "converse"
|
|
assert routes[0].user_message == "just chat"
|
|
assert routes[0].session_id == messages[0].session_id
|
|
|
|
def test_builtin_end_marks_conversation_ended(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: Literal["research", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.return_value = ResearchRoute(intent="end")
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
prompt="Classify.",
|
|
response_format=ResearchRoute,
|
|
llm=router_llm,
|
|
routes=["research"],
|
|
default_intent="converse",
|
|
)
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
return "researched"
|
|
|
|
flow = RoutedFlow()
|
|
result = flow.handle_turn("bye")
|
|
|
|
assert result == "Conversation ended."
|
|
assert flow.state.ended is True
|
|
assert flow.state.messages[-1].content == "Conversation ended."
|
|
|
|
def test_router_auto_enables_when_custom_routes_declared_and_no_explicit_config(
|
|
self,
|
|
) -> None:
|
|
"""``ConversationConfig(llm=...)`` alone wires LLM routing for custom listeners.
|
|
|
|
Users shouldn't have to pass ``router=RouterConfig()`` just to flip
|
|
the router on — declaring custom ``@listen`` handlers + giving the
|
|
config an LLM is sufficient. Only opt out by setting
|
|
``default_intents`` (legacy path).
|
|
"""
|
|
|
|
class Route(BaseModel):
|
|
intent: Literal["INTERNET_SEARCH", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.return_value = Route(intent="INTERNET_SEARCH")
|
|
|
|
@ConversationConfig(llm=router_llm) # no router= here
|
|
class AutoEnabledFlow(ConversationalFlow):
|
|
@listen("INTERNET_SEARCH")
|
|
def handle_search(self) -> str:
|
|
"""Fresh web research."""
|
|
self.append_assistant_message("searched")
|
|
return "searched"
|
|
|
|
flow = AutoEnabledFlow()
|
|
result = flow.handle_turn("research today's AI news")
|
|
|
|
assert result == "searched"
|
|
# Router LLM should have been invoked.
|
|
assert router_llm.call.call_count >= 1
|
|
|
|
def test_router_auto_enable_skipped_when_only_builtin_routes(self) -> None:
|
|
"""No custom routes → no auto-enable; falls through to converse."""
|
|
|
|
chat_llm = MagicMock()
|
|
chat_llm.call.return_value = "hi there"
|
|
|
|
@ConversationConfig(llm=chat_llm)
|
|
class NoCustomFlow(ConversationalFlow):
|
|
pass
|
|
|
|
flow = NoCustomFlow()
|
|
flow.handle_turn("hello")
|
|
|
|
assert flow.state.last_intent == "converse"
|
|
# chat_llm was used by converse_turn, not as a router.
|
|
assert chat_llm.call.call_count == 1
|
|
|
|
def test_router_auto_enable_skipped_when_default_intents_set(self) -> None:
|
|
"""Legacy ``default_intents`` opts out of router auto-enable."""
|
|
|
|
@ConversationConfig(default_intents=["search"], intent_llm="gpt-4o-mini")
|
|
class LegacyFlow(ConversationalFlow):
|
|
@listen("search")
|
|
def handle_search(self) -> str:
|
|
"""Web research."""
|
|
self.append_assistant_message("legacy-searched")
|
|
return "legacy-searched"
|
|
|
|
flow = LegacyFlow()
|
|
with patch.object(flow, "_collapse_to_outcome", return_value="search"):
|
|
result = flow.handle_turn("look it up")
|
|
|
|
# Legacy path set state.last_intent via classify_intent; auto-router did NOT
|
|
# overwrite it because default_intents short-circuits the auto-enable.
|
|
assert result == "legacy-searched"
|
|
assert flow.state.last_intent == "search"
|
|
|
|
def test_user_start_methods_run_sequentially_before_router_in_conversational_mode(
|
|
self,
|
|
) -> None:
|
|
"""Conversational flows: user ``@start`` methods finish before router fires.
|
|
|
|
Non-chat flows run ``@start`` methods in parallel via ``asyncio.gather``,
|
|
which would race with ``route_conversation`` and let the router fire
|
|
before user setup finished. In conversational mode the framework runs
|
|
them sequentially, with ``route_conversation`` last.
|
|
"""
|
|
order: list[str] = []
|
|
|
|
@ConversationConfig()
|
|
class BootstrapFlow(ConversationalFlow):
|
|
@start()
|
|
def load_profile(self) -> None:
|
|
if not self.state.session_ready:
|
|
order.append("load_profile")
|
|
self.state.session_ready = True
|
|
|
|
@start()
|
|
def attach_bus(self) -> None:
|
|
order.append("attach_bus")
|
|
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
order.append("route_turn")
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
order.append("do_work")
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = BootstrapFlow()
|
|
flow.handle_turn("turn 1")
|
|
|
|
# Both user @start methods complete before route_turn fires.
|
|
load_idx = order.index("load_profile")
|
|
attach_idx = order.index("attach_bus")
|
|
route_idx = order.index("route_turn")
|
|
assert load_idx < route_idx
|
|
assert attach_idx < route_idx
|
|
|
|
# Bootstrap gate works: load_profile only fires on the first turn.
|
|
order.clear()
|
|
flow.handle_turn("turn 2")
|
|
assert "load_profile" not in order
|
|
assert "attach_bus" in order # still fires every turn
|
|
assert "route_turn" in order
|
|
|
|
def test_subclass_can_override_conversation_start_helper(
|
|
self,
|
|
) -> None:
|
|
"""The compatibility helper remains overridable without adding a Flow node."""
|
|
|
|
bootstrap_calls: list[str] = []
|
|
|
|
@ConversationConfig()
|
|
class BootstrapFlow(ConversationalFlow):
|
|
def conversation_start(self) -> str | None:
|
|
bootstrap_calls.append("ran")
|
|
return super().conversation_start()
|
|
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = BootstrapFlow()
|
|
flow.handle_turn("hi")
|
|
|
|
assert bootstrap_calls == ["ran"]
|
|
assert "conversation_start" not in BootstrapFlow.flow_definition().methods
|
|
route_definition = BootstrapFlow.flow_definition().methods["route_conversation"]
|
|
assert route_definition.start is True
|
|
assert route_definition.router is True
|
|
assert flow.state.messages[-1].content == "worked"
|
|
|
|
def test_legacy_decorated_conversation_start_runs_once_per_turn(
|
|
self,
|
|
) -> None:
|
|
"""Legacy ``@start`` overrides are not invoked again by the router."""
|
|
|
|
bootstrap_calls: list[str] = []
|
|
|
|
@ConversationConfig()
|
|
class BootstrapFlow(ConversationalFlow):
|
|
@start()
|
|
def conversation_start(self) -> str | None:
|
|
bootstrap_calls.append("ran")
|
|
return super().conversation_start()
|
|
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = BootstrapFlow()
|
|
flow.handle_turn("hi")
|
|
|
|
assert bootstrap_calls == ["ran"]
|
|
assert flow.state.messages[-1].content == "worked"
|
|
|
|
def test_handle_turn_reruns_graph_after_prior_turn_completed(self) -> None:
|
|
"""Multi-turn must not flip ``_is_execution_resuming`` and short-circuit.
|
|
|
|
``Flow.kickoff`` with persistence enabled treats ``inputs={"id": ...}``
|
|
as a checkpoint restore, so it skips clearing ``_completed_methods``.
|
|
Without ``ConversationalFlow.kickoff`` resetting that state, turn 2+
|
|
sees every method as already-completed, short-circuits to
|
|
``_method_outputs[-1]``, and returns the previous turn's output.
|
|
"""
|
|
|
|
class Route(BaseModel):
|
|
intent: Literal["RESEARCH", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.side_effect = [
|
|
Route(intent="converse"),
|
|
Route(intent="RESEARCH"),
|
|
]
|
|
chat_llm = MagicMock()
|
|
chat_llm.call.return_value = "general help"
|
|
|
|
@ConversationConfig(
|
|
llm=chat_llm,
|
|
router=RouterConfig(
|
|
response_format=Route,
|
|
llm=router_llm,
|
|
routes=["RESEARCH"],
|
|
),
|
|
)
|
|
class DemoFlow(ConversationalFlow):
|
|
@listen("RESEARCH")
|
|
def handle_research(self) -> str:
|
|
self.append_assistant_message("fresh research")
|
|
return "fresh research"
|
|
|
|
flow = DemoFlow()
|
|
from crewai.flow.persistence import SQLiteFlowPersistence
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
flow.persistence = SQLiteFlowPersistence(
|
|
str(Path(tempfile.mkdtemp()) / "regression.db")
|
|
)
|
|
|
|
out1 = flow.handle_turn("tell me what you can do")
|
|
out2 = flow.handle_turn("now do research")
|
|
|
|
assert out1 == "general help"
|
|
assert out2 == "fresh research"
|
|
assert chat_llm.call.call_count == 1
|
|
assert router_llm.call.call_count == 2
|
|
assert flow.state.messages[-1].content == "fresh research"
|
|
assert flow._is_execution_resuming is False
|
|
|
|
def test_route_catalog_combines_docstrings_builtins_and_overrides(self) -> None:
|
|
"""Catalog precedence: route_descriptions > built-in > docstring."""
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
routes=["RESEARCH", "ORDER"],
|
|
route_descriptions={"ORDER": "explicit override for order route"},
|
|
)
|
|
)
|
|
class CatalogFlow(ConversationalFlow):
|
|
@listen("RESEARCH")
|
|
def handle_research(self) -> str:
|
|
"""Fresh web research, current news, real-time lookups."""
|
|
return "researched"
|
|
|
|
@listen("ORDER")
|
|
def handle_order(self) -> str:
|
|
"""This docstring should NOT win — override takes priority."""
|
|
return "ordered"
|
|
|
|
flow = CatalogFlow()
|
|
catalog = flow._build_route_catalog(flow.conversational_config.router)
|
|
|
|
assert catalog["RESEARCH"] == (
|
|
"Fresh web research, current news, real-time lookups."
|
|
)
|
|
assert catalog["ORDER"] == "explicit override for order route"
|
|
# Built-in routes get framework-canned descriptions.
|
|
assert "Ordinary chat" in catalog["converse"]
|
|
assert "finished" in catalog["end"]
|
|
|
|
def test_route_catalog_falls_back_to_empty_when_no_docstring(self) -> None:
|
|
@ConversationConfig(router=RouterConfig(routes=["BARE"]))
|
|
class BareFlow(ConversationalFlow):
|
|
@listen("BARE")
|
|
def handle_bare(self) -> str:
|
|
return "bare"
|
|
|
|
flow = BareFlow()
|
|
catalog = flow._build_route_catalog(flow.conversational_config.router)
|
|
|
|
assert catalog["BARE"] == ""
|
|
|
|
def test_router_messages_include_route_catalog(self) -> None:
|
|
"""The router system prompt must enumerate routes with descriptions."""
|
|
|
|
class Route(BaseModel):
|
|
intent: Literal["RESEARCH", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.return_value = Route(intent="RESEARCH")
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
prompt="A research-focused assistant.",
|
|
response_format=Route,
|
|
llm=router_llm,
|
|
routes=["RESEARCH"],
|
|
)
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("RESEARCH")
|
|
def handle_research(self) -> str:
|
|
"""Fresh web research and current news."""
|
|
self.append_assistant_message("researched")
|
|
return "researched"
|
|
|
|
flow = RoutedFlow()
|
|
flow.handle_turn("research today's AI news")
|
|
|
|
system_message = router_llm.call.call_args.kwargs["messages"][0]["content"]
|
|
assert "Routes:" in system_message
|
|
assert "- RESEARCH: Fresh web research and current news." in system_message
|
|
assert "- converse: Ordinary chat" in system_message
|
|
assert system_message.startswith("A research-focused assistant.")
|
|
|
|
def test_router_decision_persists_last_intent_and_passes_it_next_turn(
|
|
self,
|
|
) -> None:
|
|
"""Router must record its decision so the next turn's router LLM sees it."""
|
|
|
|
class Route(BaseModel):
|
|
intent: Literal["research", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.side_effect = [
|
|
Route(intent="research"),
|
|
Route(intent="converse"),
|
|
]
|
|
chat_llm = MagicMock()
|
|
chat_llm.call.return_value = "follow-up reply"
|
|
|
|
@ConversationConfig(
|
|
llm=chat_llm,
|
|
router=RouterConfig(
|
|
response_format=Route,
|
|
llm=router_llm,
|
|
routes=["research"],
|
|
),
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_assistant_message("researched")
|
|
return "researched"
|
|
|
|
flow = RoutedFlow()
|
|
|
|
flow.handle_turn("research CrewAI")
|
|
assert flow.state.last_intent == "research"
|
|
|
|
flow.handle_turn("tell me more about that")
|
|
assert flow.state.last_intent == "converse"
|
|
|
|
# Turn 2's router LLM must have seen last_intent='research' in its context.
|
|
second_call_user_content = router_llm.call.call_args_list[1].kwargs["messages"][1][
|
|
"content"
|
|
]
|
|
assert '"last_intent": "research"' in second_call_user_content
|
|
|
|
def test_custom_route_still_runs_with_builtin_routes(self) -> None:
|
|
class ResearchRoute(BaseModel):
|
|
intent: Literal["research", "converse", "end"]
|
|
|
|
router_llm = MagicMock()
|
|
router_llm.call.return_value = ResearchRoute(intent="research")
|
|
|
|
@ConversationConfig(
|
|
router=RouterConfig(
|
|
prompt="Classify.",
|
|
response_format=ResearchRoute,
|
|
llm=router_llm,
|
|
routes=["research"],
|
|
default_intent="converse",
|
|
)
|
|
)
|
|
class RoutedFlow(ConversationalFlow):
|
|
@listen("research")
|
|
def run_research(self) -> str:
|
|
self.append_agent_result("researcher", "researched", visibility="public")
|
|
return "researched"
|
|
|
|
flow = RoutedFlow()
|
|
result = flow.handle_turn("research CrewAI")
|
|
|
|
assert result == "researched"
|
|
assert flow.state.messages[-1].content == "researched"
|
|
|
|
def test_conversational_flow_auto_defaults_to_conversation_state(self) -> None:
|
|
"""``class C(Flow): conversational = True`` resolves state to ConversationState.
|
|
|
|
Pins the auto-default in ``_create_initial_state``: when the user opts
|
|
into conversational mode without an explicit ``Flow[...]`` type
|
|
parameter or ``initial_state``, state is a ``ConversationState`` with
|
|
the chat-shaped fields ready to use.
|
|
"""
|
|
|
|
class BareChat(Flow):
|
|
conversational = True
|
|
|
|
flow = BareChat()
|
|
assert isinstance(flow._state, ConversationState)
|
|
assert flow.state.messages == []
|
|
assert flow.state.current_user_message is None
|
|
assert flow.state.session_ready is False
|
|
|
|
def test_mixin_handle_turn_resolves_on_flow_subclass(self) -> None:
|
|
"""``Flow`` mixes in ``_ConversationalMixin`` — opt-in subclasses get its methods.
|
|
|
|
The conversational graph + ``handle_turn`` live on the mixin in
|
|
``crewai.experimental.conversational_mixin``; this test confirms
|
|
MRO resolution wires them onto a ``Flow`` subclass that opts in.
|
|
"""
|
|
from crewai.experimental.conversational_mixin import _ConversationalMixin
|
|
|
|
@ConversationConfig()
|
|
class MyChat(Flow):
|
|
conversational = True
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
flow = MyChat()
|
|
assert isinstance(flow, _ConversationalMixin)
|
|
assert callable(getattr(flow, "handle_turn", None))
|
|
assert callable(getattr(flow, "finalize_session_traces", None))
|
|
assert callable(getattr(flow, "append_assistant_message", None))
|
|
|
|
# Driving the mixin's handle_turn through to the listener proves
|
|
# the wiring is end-to-end, not just attribute presence.
|
|
flow.handle_turn("anything")
|
|
assert flow.state.messages[-1].content == "worked"
|
|
|
|
def test_chat_runs_repl_over_handle_turn_and_finalizes(self) -> None:
|
|
@ConversationConfig(defer_trace_finalization=False)
|
|
class MyChat(ConversationalFlow):
|
|
turns: int = 0
|
|
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.turns += 1
|
|
reply = f"worked: {self.state.current_user_message}"
|
|
self.append_assistant_message(reply)
|
|
return reply
|
|
|
|
flow = MyChat()
|
|
inputs = iter(["first", "", "second", "quit"])
|
|
prompts: list[str] = []
|
|
outputs: list[str] = []
|
|
|
|
def input_fn(prompt: str) -> str:
|
|
prompts.append(prompt)
|
|
return next(inputs)
|
|
|
|
with patch.object(flow, "finalize_session_traces") as mock_finalize:
|
|
flow.chat(
|
|
session_id="session-1",
|
|
input_fn=input_fn,
|
|
output_fn=outputs.append,
|
|
)
|
|
|
|
assert flow.turns == 2
|
|
assert prompts == ["\nYou: ", "\nYou: ", "\nYou: ", "\nYou: "]
|
|
assert outputs == [
|
|
"\nAssistant: worked: first",
|
|
"\nAssistant: worked: second",
|
|
]
|
|
mock_finalize.assert_called_once_with()
|
|
assert flow.defer_trace_finalization is False
|
|
|
|
def test_chat_stringifies_repl_output_like_conversation_helpers(self) -> None:
|
|
class RawResult:
|
|
raw = "raw assistant output"
|
|
|
|
@ConversationConfig(defer_trace_finalization=False)
|
|
class MyChat(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> RawResult:
|
|
return RawResult()
|
|
|
|
flow = MyChat()
|
|
inputs = iter(["first", "quit"])
|
|
outputs: list[str] = []
|
|
|
|
with patch.object(flow, "finalize_session_traces"):
|
|
flow.chat(
|
|
input_fn=lambda _: next(inputs),
|
|
output_fn=outputs.append,
|
|
)
|
|
|
|
assert outputs == ["\nAssistant: raw assistant output"]
|
|
|
|
def test_chat_rejects_non_conversational_flows(self) -> None:
|
|
class PlainFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "done"
|
|
|
|
flow = PlainFlow()
|
|
|
|
try:
|
|
flow.chat(input_fn=lambda _: "quit")
|
|
except ValueError as exc:
|
|
assert "conversational flows" in str(exc)
|
|
else:
|
|
raise AssertionError("Flow.chat() should reject regular flows")
|
|
|
|
def test_defer_trace_finalization_skips_per_turn_finalize(self) -> None:
|
|
"""``defer_trace_finalization = True`` suppresses per-turn ``finalize_batch``.
|
|
|
|
Without deferral, each ``handle_turn()`` ends with a trace-batch
|
|
finalize. With deferral on, the framework defers until
|
|
``finalize_session_traces()`` is called at session end.
|
|
"""
|
|
|
|
@ConversationConfig()
|
|
class DeferredFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = DeferredFlow()
|
|
flow.defer_trace_finalization = True
|
|
|
|
listener = TraceCollectionListener()
|
|
with patch.object(listener.batch_manager, "finalize_batch") as mock_finalize:
|
|
flow.handle_turn("turn 1")
|
|
flow.handle_turn("turn 2")
|
|
flow.handle_turn("turn 3")
|
|
|
|
assert mock_finalize.call_count == 0, (
|
|
"defer_trace_finalization=True must skip per-turn finalize"
|
|
)
|
|
|
|
def test_deferred_conversation_emits_one_flow_started(self) -> None:
|
|
"""Deferred conversational sessions emit one flow_started for the session."""
|
|
from crewai.events.types.flow_events import FlowStartedEvent
|
|
|
|
@ConversationConfig(defer_trace_finalization=True)
|
|
class DeferredFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = DeferredFlow()
|
|
observed_events: list[str] = []
|
|
started_events: list[FlowStartedEvent] = []
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def capture(_: Any, event: FlowStartedEvent) -> None:
|
|
observed_events.append(event.type)
|
|
started_events.append(event)
|
|
|
|
@crewai_event_bus.on(ConversationMessageAddedEvent)
|
|
def capture_message(
|
|
_: Any, event: ConversationMessageAddedEvent
|
|
) -> None:
|
|
if event.role == "user":
|
|
observed_events.append(event.type)
|
|
|
|
flow.handle_turn("turn 1")
|
|
flow.handle_turn("turn 2")
|
|
flow.handle_turn("turn 3")
|
|
crewai_event_bus.flush()
|
|
|
|
assert len(started_events) == 1, (
|
|
"deferred conversational traces should emit one session-level "
|
|
"flow_started event, not one per turn"
|
|
)
|
|
assert observed_events[0] == "flow_started"
|
|
assert observed_events[1] == "conversation_message_added"
|
|
|
|
def test_handle_turn_emits_started_and_completed_for_each_conversational_turn(
|
|
self,
|
|
) -> None:
|
|
"""Each ``handle_turn()`` emits paired turn lifecycle events."""
|
|
|
|
@ConversationConfig(defer_trace_finalization=True)
|
|
class DeferredFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = DeferredFlow()
|
|
default_session_id = flow.state.id
|
|
turn_events: list[
|
|
ConversationTurnStartedEvent | ConversationTurnCompletedEvent
|
|
] = []
|
|
|
|
original_emit = crewai_event_bus.emit
|
|
|
|
def capture_emit(source: Any, event: Any) -> Any:
|
|
if isinstance(
|
|
event, (ConversationTurnStartedEvent, ConversationTurnCompletedEvent)
|
|
):
|
|
turn_events.append(event)
|
|
return original_emit(source, event)
|
|
|
|
with patch.object(crewai_event_bus, "emit", side_effect=capture_emit):
|
|
flow.handle_turn("turn 1")
|
|
flow.handle_turn("turn 2", session_id="custom-session")
|
|
crewai_event_bus.flush()
|
|
|
|
assert [event.type for event in turn_events] == [
|
|
"conversation_turn_started",
|
|
"conversation_turn_completed",
|
|
"conversation_turn_started",
|
|
"conversation_turn_completed",
|
|
]
|
|
assert turn_events[0].session_id == default_session_id
|
|
assert turn_events[1].session_id == default_session_id
|
|
assert turn_events[2].session_id == "custom-session"
|
|
assert turn_events[3].session_id == "custom-session"
|
|
|
|
def test_handle_turn_emits_failed_instead_of_completed_when_turn_raises(
|
|
self,
|
|
) -> None:
|
|
"""Failed turns emit a terminal failure event without completion."""
|
|
|
|
@ConversationConfig(defer_trace_finalization=True)
|
|
class FailingFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
raise RuntimeError("turn exploded")
|
|
|
|
flow = FailingFlow()
|
|
turn_events: list[
|
|
ConversationTurnStartedEvent
|
|
| ConversationTurnCompletedEvent
|
|
| ConversationTurnFailedEvent
|
|
] = []
|
|
handled_failed_events: list[ConversationTurnFailedEvent] = []
|
|
original_emit = crewai_event_bus.emit
|
|
|
|
def capture_emit(source: Any, event: Any) -> Any:
|
|
if isinstance(
|
|
event,
|
|
(
|
|
ConversationTurnStartedEvent,
|
|
ConversationTurnCompletedEvent,
|
|
ConversationTurnFailedEvent,
|
|
),
|
|
):
|
|
turn_events.append(event)
|
|
return original_emit(source, event)
|
|
|
|
with (
|
|
crewai_event_bus.scoped_handlers(),
|
|
patch.object(crewai_event_bus, "emit", side_effect=capture_emit),
|
|
):
|
|
|
|
@crewai_event_bus.on(ConversationTurnFailedEvent)
|
|
def capture_failed(
|
|
_: Any, event: ConversationTurnFailedEvent
|
|
) -> None:
|
|
handled_failed_events.append(event)
|
|
|
|
with pytest.raises(RuntimeError, match="turn exploded"):
|
|
flow.handle_turn("turn 1")
|
|
|
|
assert [event.type for event in turn_events] == [
|
|
"conversation_turn_started",
|
|
"conversation_turn_failed",
|
|
]
|
|
assert turn_events[0].session_id == flow.state.id
|
|
failed_event = turn_events[1]
|
|
assert isinstance(failed_event, ConversationTurnFailedEvent)
|
|
assert failed_event.session_id == flow.state.id
|
|
assert str(failed_event.error) == "turn exploded"
|
|
assert handled_failed_events == [failed_event]
|
|
|
|
def test_conversation_turn_completed_tracks_feature_usage(self) -> None:
|
|
"""Completed conversation turns count conversational Flow usage."""
|
|
from crewai.events.event_listener import event_listener
|
|
|
|
@ConversationConfig(defer_trace_finalization=True)
|
|
class DeferredFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = DeferredFlow()
|
|
|
|
with (
|
|
crewai_event_bus.scoped_handlers(),
|
|
patch.object(
|
|
event_listener._telemetry,
|
|
"feature_usage_span",
|
|
) as feature_usage_span,
|
|
):
|
|
event_listener.setup_listeners(crewai_event_bus)
|
|
flow.handle_turn("turn 1")
|
|
|
|
feature_usage_span.assert_any_call("flow:conversation_turn")
|
|
|
|
def test_route_event_uses_no_message_index_for_empty_transcript(self) -> None:
|
|
"""Route events do not reference index zero when no message exists."""
|
|
|
|
@ConversationConfig()
|
|
class DeferredFlow(ConversationalFlow):
|
|
pass
|
|
|
|
flow = DeferredFlow()
|
|
route_events: list[ConversationRouteSelectedEvent] = []
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(ConversationRouteSelectedEvent)
|
|
def capture(_: Any, event: ConversationRouteSelectedEvent) -> None:
|
|
route_events.append(event)
|
|
|
|
flow._emit_conversation_route_selected("converse")
|
|
crewai_event_bus.flush()
|
|
|
|
assert len(route_events) == 1
|
|
assert route_events[0].message_index is None
|
|
|
|
def test_finalize_session_traces_emits_finished_and_finalizes_batch(self) -> None:
|
|
"""``finalize_session_traces()`` emits one ``FlowFinishedEvent`` + one ``finalize_batch``.
|
|
|
|
Pairs with the deferral above: after N turns with deferral on, a
|
|
single ``finalize_session_traces()`` closes the whole session as
|
|
one trace batch with one terminal event.
|
|
"""
|
|
from crewai.events.types.flow_events import FlowFinishedEvent
|
|
|
|
@ConversationConfig()
|
|
class DeferredFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = DeferredFlow()
|
|
flow.defer_trace_finalization = True
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.batch_owner_type = "flow"
|
|
listener.first_time_handler.is_first_time = False
|
|
|
|
finished_events: list[FlowFinishedEvent] = []
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def capture(_: Any, event: FlowFinishedEvent) -> None:
|
|
finished_events.append(event)
|
|
|
|
with patch.object(
|
|
listener.batch_manager, "finalize_batch"
|
|
) as mock_finalize:
|
|
flow.handle_turn("turn 1")
|
|
crewai_event_bus.flush()
|
|
flow.handle_turn("turn 2")
|
|
crewai_event_bus.flush()
|
|
# No flow_finished or finalize_batch yet — deferred.
|
|
assert finished_events == []
|
|
assert mock_finalize.call_count == 0
|
|
|
|
flow.finalize_session_traces()
|
|
crewai_event_bus.flush()
|
|
|
|
assert len(finished_events) == 1, (
|
|
"finalize_session_traces must emit exactly one FlowFinishedEvent"
|
|
)
|
|
assert mock_finalize.call_count == 1, (
|
|
"finalize_session_traces must finalize the trace batch once"
|
|
)
|
|
|
|
def test_deferred_resume_skips_per_resume_flow_finished_event(self) -> None:
|
|
"""Deferred sessions do not emit terminal events while resuming."""
|
|
from crewai.events.types.flow_events import FlowFinishedEvent
|
|
from crewai.flow.async_feedback.types import PendingFeedbackContext
|
|
|
|
class DeferredResumeFlow(Flow[ChatState]):
|
|
defer_trace_finalization = True
|
|
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "started"
|
|
|
|
flow = DeferredResumeFlow()
|
|
flow._pending_feedback_context = PendingFeedbackContext(
|
|
flow_id=flow.flow_id,
|
|
flow_class="DeferredResumeFlow",
|
|
method_name="begin",
|
|
method_output="started",
|
|
message="Review",
|
|
)
|
|
|
|
finished_events: list[FlowFinishedEvent] = []
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def capture(_: Any, event: FlowFinishedEvent) -> None:
|
|
finished_events.append(event)
|
|
|
|
flow.resume("approved")
|
|
crewai_event_bus.flush()
|
|
|
|
assert finished_events == []
|
|
|
|
def test_finalize_session_traces_restores_event_scope(self, capsys) -> None:
|
|
"""No ``empty scope stack`` warning when deferred ``flow_finished`` fires.
|
|
|
|
The first turn's ``flow_started`` event id is stashed on the flow
|
|
so ``finalize_session_traces`` can restore the scope before emitting
|
|
``flow_finished``. Without this, the event bus prints
|
|
``Warning: Ending event 'flow_finished' emitted with empty scope stack``.
|
|
"""
|
|
|
|
@ConversationConfig()
|
|
class DeferredFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = DeferredFlow()
|
|
flow.defer_trace_finalization = True
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.batch_owner_type = "flow"
|
|
listener.first_time_handler.is_first_time = False
|
|
|
|
with patch.object(listener.batch_manager, "finalize_batch"):
|
|
flow.handle_turn("hi")
|
|
flow.finalize_session_traces()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Missing starting event" not in (captured.out + captured.err), (
|
|
"finalize_session_traces should restore the flow_started scope so "
|
|
"the event bus pairs flow_finished with its opener"
|
|
)
|
|
|
|
def test_finalize_session_traces_is_noop_when_not_deferred(self) -> None:
|
|
"""Without deferral, ``finalize_session_traces()`` must not re-emit.
|
|
|
|
Each per-turn ``handle_turn()`` already emits its own
|
|
``flow_finished``; a defensive ``try/finally`` call to
|
|
``finalize_session_traces()`` at session end must not emit a second,
|
|
unpaired session-end event (which would confuse tracing).
|
|
"""
|
|
from crewai.events.types.flow_events import FlowFinishedEvent
|
|
|
|
@ConversationConfig(defer_trace_finalization=False)
|
|
class PlainFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = PlainFlow() # finalization NOT deferred
|
|
|
|
# A non-deferred turn closes itself (no flow_started stashed for later).
|
|
flow.handle_turn("turn 1")
|
|
crewai_event_bus.flush()
|
|
assert getattr(flow, "_deferred_flow_started_event_id", None) is None
|
|
|
|
# Capture only what finalize_session_traces emits.
|
|
finished_events: list[FlowFinishedEvent] = []
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def capture(_: Any, event: FlowFinishedEvent) -> None:
|
|
finished_events.append(event)
|
|
|
|
flow.finalize_session_traces()
|
|
crewai_event_bus.flush()
|
|
|
|
assert finished_events == [], (
|
|
"finalize_session_traces must be a no-op when finalization was not "
|
|
"deferred — it should not emit a duplicate flow_finished"
|
|
)
|
|
|
|
|
|
class TestHandleTurnReplyFallback:
|
|
"""Regression tests for EPD-181: ``handle_turn()`` decided "did the
|
|
handler append its reply?" by comparing assistant-message counts. A
|
|
handler that appends its reply AND trims history to a cap left the count
|
|
unchanged, so the fallback appended the reply a second time — every turn,
|
|
once trimming engaged. The check now uses an explicit appended-this-turn
|
|
flag.
|
|
"""
|
|
|
|
MAX_MESSAGES = 4
|
|
|
|
def _make_bot(self) -> ConversationalFlow:
|
|
max_messages = self.MAX_MESSAGES
|
|
|
|
class EchoBot(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "ECHO"
|
|
|
|
@listen("ECHO")
|
|
def echo(self) -> str:
|
|
reply = f"echo: {self.state.current_user_message or ''}"
|
|
self.append_assistant_message(reply) # handler DOES append
|
|
if len(self.state.messages) > max_messages: # ...and trims
|
|
self.state.messages = self.state.messages[-max_messages:]
|
|
return reply
|
|
|
|
return EchoBot()
|
|
|
|
def test_no_duplicate_reply_when_handler_trims_history(self) -> None:
|
|
bot = self._make_bot()
|
|
for i in range(1, 5):
|
|
bot.handle_turn(f"message {i}")
|
|
contents = [message.content for message in bot.state.messages]
|
|
assert len(contents) == len(set(contents)), (
|
|
f"duplicate reply on turn {i}: {contents}"
|
|
)
|
|
|
|
# The capped window holds the last two full turns, in order.
|
|
assert [message.content for message in bot.state.messages] == [
|
|
"message 3",
|
|
"echo: message 3",
|
|
"message 4",
|
|
"echo: message 4",
|
|
]
|
|
|
|
def test_fallback_still_appends_when_handler_does_not_reply(self) -> None:
|
|
class SilentBot(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "WORK"
|
|
|
|
@listen("WORK")
|
|
def work(self) -> str:
|
|
return "computed reply" # returns without appending
|
|
|
|
bot = SilentBot()
|
|
bot.handle_turn("hello")
|
|
|
|
assistant_messages = [
|
|
message.content
|
|
for message in bot.state.messages
|
|
if message.role == "assistant"
|
|
]
|
|
assert assistant_messages == ["computed reply"]
|
|
|
|
|
|
class TestFalsyRouteTurnFallback:
|
|
"""A falsy ``route_turn()`` must never replay a previous turn's intent.
|
|
|
|
Regression tests for EPD-176: an overridden ``route_turn()`` returning
|
|
``None`` on an unhandled input used to silently reuse the sticky
|
|
``state.last_intent`` from the *previous* turn, running the wrong handler
|
|
with no error or warning.
|
|
"""
|
|
|
|
def test_falsy_route_turn_does_not_replay_previous_turns_intent(self) -> None:
|
|
ran: list[str] = []
|
|
|
|
class Bot(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
message = context.get("current_user_message") or ""
|
|
if "hello" in message.lower():
|
|
return "GREETING"
|
|
return None # unhandled input -> falsy return
|
|
|
|
@listen("GREETING")
|
|
def greeting(self) -> str:
|
|
ran.append("GREETING")
|
|
reply = "Hi! I only do greetings."
|
|
self.append_assistant_message(reply)
|
|
return reply
|
|
|
|
@listen("WEATHER")
|
|
def weather(self) -> str:
|
|
ran.append("WEATHER")
|
|
reply = "It is sunny."
|
|
self.append_assistant_message(reply)
|
|
return reply
|
|
|
|
flow = Bot()
|
|
flow.handle_turn("hello there")
|
|
assert ran == ["GREETING"]
|
|
assert flow.state.last_intent == "GREETING"
|
|
|
|
flow.handle_turn("what is the meaning of life?")
|
|
assert ran == ["GREETING"], (
|
|
"an unhandled turn must not re-run the previous turn's handler"
|
|
)
|
|
# With no routing decision the turn falls through to the built-in
|
|
# 'converse' default instead of replaying the stale intent.
|
|
assert flow.state.last_intent == "converse"
|
|
assert flow.state.messages[-1].content != "Hi! I only do greetings."
|
|
|
|
def test_stale_intent_ignored_but_route_selected_event_still_emitted(
|
|
self,
|
|
) -> None:
|
|
class Bot(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
message = context.get("current_user_message") or ""
|
|
return "work" if "work" in message else None
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = Bot()
|
|
routes: list[ConversationRouteSelectedEvent] = []
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(ConversationRouteSelectedEvent)
|
|
def capture(_: Any, event: ConversationRouteSelectedEvent) -> None:
|
|
routes.append(event)
|
|
|
|
flow.handle_turn("work please")
|
|
flow.handle_turn("something unrelated")
|
|
crewai_event_bus.flush()
|
|
|
|
assert [event.route for event in routes] == ["work", "converse"]
|
|
# The fallback decision still reports the prior intent for visibility.
|
|
assert routes[1].previous_intent == "work"
|
|
|
|
def test_fresh_intent_classified_this_turn_still_routes(self) -> None:
|
|
"""The legacy ``default_intents`` path classifies per turn and must
|
|
keep routing on the freshly classified intent — including when the
|
|
intent changes between turns."""
|
|
ran: list[str] = []
|
|
|
|
@ConversationConfig(
|
|
default_intents=["search", "weather"], intent_llm="gpt-4o-mini"
|
|
)
|
|
class LegacyFlow(ConversationalFlow):
|
|
@listen("search")
|
|
def handle_search(self) -> str:
|
|
ran.append("search")
|
|
self.append_assistant_message("searched")
|
|
return "searched"
|
|
|
|
@listen("weather")
|
|
def handle_weather(self) -> str:
|
|
ran.append("weather")
|
|
self.append_assistant_message("sunny")
|
|
return "sunny"
|
|
|
|
flow = LegacyFlow()
|
|
with patch.object(
|
|
flow, "_collapse_to_outcome", side_effect=["search", "weather"]
|
|
):
|
|
flow.handle_turn("look up crewai")
|
|
flow.handle_turn("how is the weather?")
|
|
|
|
assert ran == ["search", "weather"]
|
|
assert flow.state.last_intent == "weather"
|
|
|
|
|
|
class TestFlowTracingWhenSuppressed:
|
|
def test_flow_started_emitted_when_panel_events_suppressed(self) -> None:
|
|
class QuietFlow(Flow[ChatState]):
|
|
suppress_flow_events = True
|
|
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "ok"
|
|
|
|
started: list[str] = []
|
|
original_emit = crewai_event_bus.emit
|
|
|
|
def track_emit(source: Any, event: Any, *args: Any, **kwargs: Any) -> Any:
|
|
if isinstance(event, FlowStartedEvent):
|
|
started.append(event.flow_name)
|
|
return original_emit(source, event, *args, **kwargs)
|
|
|
|
with patch.object(crewai_event_bus, "emit", side_effect=track_emit):
|
|
QuietFlow().kickoff()
|
|
|
|
assert started == ["QuietFlow"]
|
|
|
|
def test_method_execution_suppressed_when_flow_events_suppressed(self) -> None:
|
|
"""``suppress_flow_events=True`` silences MethodExecution events so
|
|
infrastructure flows (AgentExecutor, memory) don't emit one trace span
|
|
per internal control-flow method."""
|
|
|
|
class QuietFlow(Flow[ChatState]):
|
|
suppress_flow_events = True
|
|
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "ok"
|
|
|
|
started: list[str] = []
|
|
finished: list[str] = []
|
|
original_emit = crewai_event_bus.emit
|
|
|
|
def track_emit(source: Any, event: Any, *args: Any, **kwargs: Any) -> Any:
|
|
if isinstance(event, MethodExecutionStartedEvent):
|
|
started.append(event.method_name)
|
|
if isinstance(event, MethodExecutionFinishedEvent):
|
|
finished.append(event.method_name)
|
|
return original_emit(source, event, *args, **kwargs)
|
|
|
|
with patch.object(crewai_event_bus, "emit", side_effect=track_emit):
|
|
QuietFlow().kickoff()
|
|
|
|
assert started == []
|
|
assert finished == []
|
|
|
|
def test_llm_action_inside_flow_claims_flow_trace_batch(self) -> None:
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.current_batch = None
|
|
listener.batch_manager.batch_owner_type = None
|
|
listener.batch_manager.batch_owner_id = None
|
|
|
|
flow_id_token = current_flow_id.set("flow-test-id")
|
|
flow_name_token = current_flow_name.set("DemoSupportFlow")
|
|
try:
|
|
event = LLMCallStartedEvent(
|
|
model="gpt-4o-mini",
|
|
messages=[],
|
|
call_id="call-test",
|
|
)
|
|
listener._handle_action_event("llm_call_started", object(), event)
|
|
finally:
|
|
current_flow_id.reset(flow_id_token)
|
|
current_flow_name.reset(flow_name_token)
|
|
|
|
assert listener.batch_manager.batch_owner_type == "flow"
|
|
assert listener.batch_manager.batch_owner_id == "flow-test-id"
|
|
assert (
|
|
listener.batch_manager.current_batch.execution_metadata["execution_type"]
|
|
== "flow"
|
|
)
|
|
assert (
|
|
listener.batch_manager.current_batch.execution_metadata["flow_name"]
|
|
== "DemoSupportFlow"
|
|
)
|
|
|
|
|
|
class TestDeferTraceFinalization:
|
|
def test_bare_conversational_flow_defers_by_default(self) -> None:
|
|
class BareChat(ConversationalFlow):
|
|
pass
|
|
|
|
assert BareChat()._should_defer_trace_finalization() is True
|
|
|
|
def test_conversation_config_drives_defer_flag(self) -> None:
|
|
"""``ConversationConfig(defer_trace_finalization=...)`` controls whether
|
|
a conversational subclass defers per-turn trace finalization."""
|
|
|
|
@ConversationConfig(defer_trace_finalization=True)
|
|
class DeferOn(ConversationalFlow):
|
|
pass
|
|
|
|
@ConversationConfig(defer_trace_finalization=False)
|
|
class DeferOff(ConversationalFlow):
|
|
pass
|
|
|
|
assert DeferOn()._should_defer_trace_finalization() is True
|
|
assert DeferOff()._should_defer_trace_finalization() is False
|
|
|
|
|
|
|
|
class TestDeferredFlowLifecycleEvents:
|
|
def test_flow_finished_without_flow_started_warns(self, capsys) -> None:
|
|
from crewai.events.event_bus import crewai_event_bus
|
|
from crewai.events.event_context import restore_event_scope
|
|
from crewai.events.types.flow_events import FlowFinishedEvent
|
|
|
|
class BareFlow(Flow[ChatState]):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "ok"
|
|
|
|
restore_event_scope(())
|
|
flow = BareFlow()
|
|
crewai_event_bus.emit(
|
|
flow,
|
|
FlowFinishedEvent(
|
|
type="flow_finished",
|
|
flow_name="BareFlow",
|
|
result="ok",
|
|
state={},
|
|
),
|
|
)
|
|
captured = capsys.readouterr().out
|
|
assert "flow_finished" in captured
|
|
assert "Missing starting event" in captured
|
|
|
|
def test_finalize_batch_is_idempotent(self) -> None:
|
|
from crewai.events.listeners.tracing.trace_batch_manager import TraceBatchManager
|
|
|
|
with patch(
|
|
"crewai.events.listeners.tracing.trace_batch_manager.is_tracing_enabled_in_context",
|
|
return_value=True,
|
|
):
|
|
bm = TraceBatchManager()
|
|
bm.current_batch = bm.initialize_batch(
|
|
user_context={"privacy_level": "standard"},
|
|
execution_metadata={"execution_type": "flow", "flow_name": "ChatFlow"},
|
|
)
|
|
bm.trace_batch_id = "batch-idempotent"
|
|
bm.backend_initialized = True
|
|
|
|
with (
|
|
patch.object(
|
|
bm.plus_api,
|
|
"send_trace_events",
|
|
return_value=MagicMock(status_code=200),
|
|
),
|
|
patch.object(
|
|
bm.plus_api,
|
|
"finalize_trace_batch",
|
|
return_value=MagicMock(status_code=200, json=MagicMock(return_value={})),
|
|
) as mock_finalize_api,
|
|
):
|
|
bm.finalize_batch()
|
|
bm.finalize_batch()
|
|
|
|
assert mock_finalize_api.call_count == 1
|
|
assert bm._batch_finalized is True
|
|
|
|
def test_finalize_session_traces_is_idempotent(self) -> None:
|
|
"""Calling ``finalize_session_traces()`` twice emits flow_finished once.
|
|
|
|
The stashed ``_deferred_flow_started_event_id`` is cleared after the
|
|
first call, so a second call (e.g. a defensive ``try/finally``) does
|
|
not re-emit a session-end event.
|
|
"""
|
|
from crewai.events.types.flow_events import FlowFinishedEvent
|
|
|
|
@ConversationConfig(defer_trace_finalization=True)
|
|
class DeferredFlow(ConversationalFlow):
|
|
def route_turn(self, context: dict[str, Any]) -> str | None:
|
|
return "work"
|
|
|
|
@listen("work")
|
|
def do_work(self) -> str:
|
|
self.append_assistant_message("worked")
|
|
return "worked"
|
|
|
|
flow = DeferredFlow()
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.batch_owner_type = "flow"
|
|
listener.first_time_handler.is_first_time = False
|
|
|
|
finished: list[FlowFinishedEvent] = []
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def capture(_: Any, event: FlowFinishedEvent) -> None:
|
|
finished.append(event)
|
|
|
|
with patch.object(listener.batch_manager, "finalize_batch"):
|
|
flow.handle_turn("hi")
|
|
crewai_event_bus.flush()
|
|
flow.finalize_session_traces()
|
|
flow.finalize_session_traces() # second call must be a no-op
|
|
crewai_event_bus.flush()
|
|
|
|
assert len(finished) == 1, (
|
|
"finalize_session_traces must emit flow_finished exactly once, even "
|
|
"when called more than once"
|
|
)
|
|
|
|
def test_sigint_skips_deferred_session_batch(self) -> None:
|
|
from crewai.events.listeners.tracing.trace_batch_manager import TraceBatch
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.current_batch = TraceBatch()
|
|
listener.batch_manager.defer_session_finalization = True
|
|
|
|
with patch.object(listener.batch_manager, "finalize_batch") as mock_finalize:
|
|
if listener.batch_manager.is_batch_initialized():
|
|
if not listener.batch_manager.defer_session_finalization:
|
|
listener.batch_manager.finalize_batch()
|
|
mock_finalize.assert_not_called()
|
|
|
|
def test_deferred_flow_kickoff_marks_trace_manager_session_deferred(
|
|
self,
|
|
) -> None:
|
|
class DeferredTraceFlow(Flow[ChatState]):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "done"
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.defer_session_finalization = False
|
|
|
|
flow = DeferredTraceFlow()
|
|
flow.defer_trace_finalization = True
|
|
|
|
with patch.object(listener.batch_manager, "finalize_batch"):
|
|
flow.kickoff()
|
|
|
|
assert listener.batch_manager.defer_session_finalization is True
|
|
|
|
flow.finalize_session_traces()
|
|
|
|
assert listener.batch_manager.defer_session_finalization is False
|
|
|
|
def test_non_deferred_flow_kickoff_clears_stale_trace_manager_flag(
|
|
self,
|
|
) -> None:
|
|
class PlainTraceFlow(Flow[ChatState]):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "done"
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.defer_session_finalization = True
|
|
|
|
PlainTraceFlow().kickoff()
|
|
|
|
assert listener.batch_manager.defer_session_finalization is False
|
|
|
|
|
|
class TestNestedCrewTracing:
|
|
def test_is_inside_active_flow_context_when_kickoff_running(self) -> None:
|
|
from crewai.events.listeners.tracing.trace_listener import (
|
|
TraceCollectionListener,
|
|
)
|
|
from crewai.flow.flow_context import current_flow_id
|
|
|
|
assert TraceCollectionListener._is_inside_active_flow_context() is False
|
|
token = current_flow_id.set("parent-flow-id")
|
|
try:
|
|
assert TraceCollectionListener._is_inside_active_flow_context() is True
|
|
finally:
|
|
current_flow_id.reset(token)
|
|
|
|
def test_nested_crew_completion_skips_finalize(self) -> None:
|
|
from crewai.events.listeners.tracing.trace_listener import (
|
|
TraceCollectionListener,
|
|
)
|
|
from crewai.flow.flow_context import current_flow_id
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.batch_owner_type = "crew"
|
|
|
|
token = current_flow_id.set("parent-flow-id")
|
|
try:
|
|
with patch.object(listener.batch_manager, "finalize_batch") as mock_finalize:
|
|
if listener._nested_in_flow_execution():
|
|
pass
|
|
elif listener.batch_manager.batch_owner_type == "crew":
|
|
listener.batch_manager.finalize_batch()
|
|
mock_finalize.assert_not_called()
|
|
finally:
|
|
current_flow_id.reset(token)
|
|
|
|
def test_flow_owned_batch_skips_finalize_without_flow_context(self) -> None:
|
|
from crewai.events.listeners.tracing.trace_listener import (
|
|
TraceCollectionListener,
|
|
)
|
|
from crewai.events.listeners.tracing.trace_batch_manager import TraceBatch
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.batch_owner_type = "flow"
|
|
listener.batch_manager.current_batch = TraceBatch(
|
|
execution_metadata={"execution_type": "flow", "flow_name": "Demo"},
|
|
)
|
|
|
|
with patch.object(listener.batch_manager, "finalize_batch") as mock_finalize:
|
|
if listener._nested_in_flow_execution():
|
|
pass
|
|
elif listener.batch_manager.batch_owner_type == "crew":
|
|
listener.batch_manager.finalize_batch()
|
|
mock_finalize.assert_not_called()
|
|
|
|
def test_lazy_flow_batch_from_context_preserves_deferred_parent(self) -> None:
|
|
from crewai.events.listeners.tracing.trace_listener import (
|
|
TraceCollectionListener,
|
|
)
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.current_batch = None
|
|
listener.batch_manager.batch_owner_type = None
|
|
listener.batch_manager.batch_owner_id = None
|
|
listener.batch_manager.defer_session_finalization = False
|
|
listener.batch_manager.event_buffer.clear()
|
|
|
|
flow_id_token = current_flow_id.set("parent-flow-id")
|
|
flow_name_token = current_flow_name.set("ParentChatFlow")
|
|
defer_token = current_flow_defer_trace_finalization.set(True)
|
|
try:
|
|
initialized = listener._try_initialize_flow_batch_from_context(
|
|
type("Event", (), {"timestamp": None})()
|
|
)
|
|
|
|
assert initialized is True
|
|
assert listener.batch_manager.batch_owner_type == "flow"
|
|
assert listener.batch_manager.batch_owner_id == "parent-flow-id"
|
|
assert listener.batch_manager.defer_session_finalization is True
|
|
assert listener.batch_manager.current_batch is not None
|
|
assert (
|
|
listener.batch_manager.current_batch.execution_metadata[
|
|
"execution_type"
|
|
]
|
|
== "flow"
|
|
)
|
|
assert (
|
|
listener.batch_manager.current_batch.execution_metadata["flow_name"]
|
|
== "ParentChatFlow"
|
|
)
|
|
finally:
|
|
current_flow_defer_trace_finalization.reset(defer_token)
|
|
current_flow_name.reset(flow_name_token)
|
|
current_flow_id.reset(flow_id_token)
|
|
listener.batch_manager.current_batch = None
|
|
listener.batch_manager.batch_owner_type = None
|
|
listener.batch_manager.batch_owner_id = None
|
|
listener.batch_manager.trace_batch_id = None
|
|
listener.batch_manager.defer_session_finalization = False
|
|
listener.batch_manager.event_buffer.clear()
|
|
|
|
def test_nested_agent_executor_flow_does_not_finalize_parent_batch(
|
|
self,
|
|
) -> None:
|
|
from crewai import Agent, Crew, Task
|
|
from crewai.llms.base_llm import BaseLLM
|
|
|
|
class StaticLLM(BaseLLM):
|
|
def __init__(self) -> None:
|
|
super().__init__(model="debug-static-llm", provider="debug")
|
|
|
|
def call(
|
|
self,
|
|
messages: Any,
|
|
tools: Any = None,
|
|
callbacks: Any = None,
|
|
available_functions: Any = None,
|
|
from_task: Any = None,
|
|
from_agent: Any = None,
|
|
response_model: Any = None,
|
|
) -> str:
|
|
return (
|
|
"Thought: I can answer directly.\n"
|
|
"Final Answer: nested crew result"
|
|
)
|
|
|
|
class NestedCrewFlow(Flow[ChatState]):
|
|
defer_trace_finalization = True
|
|
tracing = True
|
|
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "run_nested_crew"
|
|
|
|
@listen(begin)
|
|
def run_nested_crew(self, _: str) -> str:
|
|
agent = Agent(
|
|
role="Debug Agent",
|
|
goal="Return a short deterministic result",
|
|
backstory="Used only for trace finalization debugging.",
|
|
llm=StaticLLM(),
|
|
verbose=False,
|
|
)
|
|
task = Task(
|
|
description="Return the deterministic nested crew result.",
|
|
expected_output="nested crew result",
|
|
agent=agent,
|
|
)
|
|
return Crew(agents=[agent], tasks=[task], verbose=False).kickoff().raw
|
|
|
|
listener = TraceCollectionListener()
|
|
listener.batch_manager.current_batch = None
|
|
listener.batch_manager.batch_owner_type = None
|
|
listener.batch_manager.batch_owner_id = None
|
|
listener.batch_manager.trace_batch_id = None
|
|
listener.batch_manager.defer_session_finalization = False
|
|
listener.batch_manager.event_buffer.clear()
|
|
listener.first_time_handler.is_first_time = False
|
|
|
|
def initialize_backend_batch(*_: Any, **__: Any) -> None:
|
|
listener.batch_manager.trace_batch_id = "debug-trace-batch"
|
|
|
|
flow = NestedCrewFlow()
|
|
|
|
with (
|
|
patch.object(
|
|
listener.batch_manager,
|
|
"_initialize_backend_batch",
|
|
side_effect=initialize_backend_batch,
|
|
),
|
|
patch.object(listener.batch_manager, "finalize_batch") as mock_finalize,
|
|
):
|
|
flow.kickoff()
|
|
crewai_event_bus.flush()
|
|
flow.kickoff()
|
|
crewai_event_bus.flush()
|
|
|
|
assert mock_finalize.call_count == 0, (
|
|
"nested AgentExecutor flows inside a deferred parent Flow must "
|
|
"not finalize the parent trace batch"
|
|
)
|
|
|
|
|
|
class TestConversationalOptIn:
|
|
"""``@ConversationConfig`` opts a Flow into conversational mode."""
|
|
|
|
def test_decorator_alone_enables_conversational_mode(self) -> None:
|
|
@ConversationConfig(llm="gpt-4o-mini")
|
|
class DecoratedFlow(Flow[ConversationState]):
|
|
@listen("order")
|
|
def handle_order(self) -> str:
|
|
"""Order status questions."""
|
|
return "on the way"
|
|
|
|
definition = DecoratedFlow.flow_definition()
|
|
|
|
assert definition.conversational is not None
|
|
assert definition.conversational.enabled is True
|
|
assert definition.conversational.llm == "gpt-4o-mini"
|
|
|
|
def test_decorator_alone_registers_the_builtin_methods(self) -> None:
|
|
@ConversationConfig()
|
|
class DecoratedFlow(Flow[ConversationState]):
|
|
pass
|
|
|
|
methods = DecoratedFlow.flow_definition().methods
|
|
|
|
assert "route_conversation" in methods
|
|
assert methods["route_conversation"].start is True
|
|
assert methods["route_conversation"].router is True
|
|
assert "converse_turn" in methods
|
|
assert "end_conversation" in methods
|
|
|
|
def test_decorator_alone_runs_a_turn(self) -> None:
|
|
@ConversationConfig()
|
|
class DecoratedFlow(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> str:
|
|
return "handled"
|
|
|
|
flow = DecoratedFlow()
|
|
|
|
assert flow.handle_turn("hello") == "handled"
|
|
assert [(m.role, m.content) for m in flow.state.messages] == [
|
|
("user", "hello"),
|
|
("assistant", "handled"),
|
|
]
|
|
|
|
def test_decorator_alone_defers_trace_finalization(self) -> None:
|
|
@ConversationConfig()
|
|
class DecoratedFlow(Flow[ConversationState]):
|
|
pass
|
|
|
|
assert DecoratedFlow()._should_defer_trace_finalization() is True
|
|
|
|
def test_explicit_flag_without_a_config_still_opts_in(self) -> None:
|
|
class FlagOnlyFlow(Flow[ConversationState]):
|
|
conversational = True
|
|
|
|
definition = FlagOnlyFlow.flow_definition()
|
|
|
|
assert definition.conversational is not None
|
|
assert definition.conversational.enabled is True
|
|
assert FlagOnlyFlow()._is_conversational_enabled() is True
|
|
|
|
def test_flow_with_neither_stays_non_conversational(self) -> None:
|
|
class PlainFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "begin"
|
|
|
|
definition = PlainFlow.flow_definition()
|
|
|
|
assert definition.conversational is None
|
|
assert set(definition.methods) == {"begin"}
|
|
assert PlainFlow.conversational is False
|
|
|
|
def test_decorator_does_not_leak_the_flag_onto_other_flows(self) -> None:
|
|
@ConversationConfig()
|
|
class DecoratedFlow(Flow[ConversationState]):
|
|
pass
|
|
|
|
class LaterPlainFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "begin"
|
|
|
|
assert DecoratedFlow.conversational is True
|
|
assert LaterPlainFlow.conversational is False
|
|
assert Flow.conversational is False
|
|
assert LaterPlainFlow.flow_definition().conversational is None
|
|
|
|
|
|
class TestHandleTurnGuard:
|
|
"""``handle_turn`` fails loudly on a non-conversational Flow."""
|
|
|
|
def test_handle_turn_rejects_non_conversational_flows(self) -> None:
|
|
class PlainFlow(Flow[ConversationState]):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "begin"
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="Flow.handle_turn\\(\\) is only available on conversational flows",
|
|
):
|
|
PlainFlow().handle_turn("hello")
|
|
|
|
def test_handle_turn_guard_matches_chat_and_stream_turn(self) -> None:
|
|
class PlainFlow(Flow[ConversationState]):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "begin"
|
|
|
|
flow = PlainFlow()
|
|
|
|
for call in (
|
|
lambda: flow.handle_turn("hi"),
|
|
lambda: flow.stream_turn("hi"),
|
|
flow.chat,
|
|
):
|
|
with pytest.raises(
|
|
ValueError, match="only available on conversational flows"
|
|
):
|
|
call()
|
|
|
|
def test_guard_does_not_fire_on_a_conversational_flow(self) -> None:
|
|
@ConversationConfig()
|
|
class DecoratedFlow(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> str:
|
|
return "ok"
|
|
|
|
assert DecoratedFlow().handle_turn("hi") == "ok"
|
|
|
|
|
|
_MIXIN = "crewai.experimental.conversational_mixin:_ConversationalMixin"
|
|
|
|
|
|
class DeclaredChatState(ConversationState):
|
|
"""Module-level so ``state.ref`` can import it; a locals-scoped class cannot."""
|
|
|
|
ticket_id: str | None = None
|
|
|
|
|
|
class _ScriptedLLM(BaseLLM):
|
|
"""Fake LLM returning queued responses; records the messages it saw."""
|
|
|
|
def __init__(self, responses: list[str] | None = None) -> None:
|
|
super().__init__(model="fake")
|
|
object.__setattr__(self, "_responses", list(responses or []))
|
|
object.__setattr__(self, "seen", [])
|
|
|
|
def call(self, messages, **kwargs) -> str: # type: ignore[no-untyped-def]
|
|
self.seen.append(messages)
|
|
return self._responses.pop(0) if self._responses else "fallback"
|
|
|
|
def supports_function_calling(self) -> bool:
|
|
return False
|
|
|
|
def supports_stop_words(self) -> bool:
|
|
return False
|
|
|
|
def get_context_window_size(self) -> int:
|
|
return 8192
|
|
|
|
|
|
def _conversational_declaration(**overrides: Any) -> dict[str, Any]:
|
|
"""A declaration naming the built-in methods explicitly.
|
|
|
|
Synthesizing these entries from ``conversational.enabled`` is a follow-up;
|
|
until then a declaration has to name them.
|
|
"""
|
|
declaration: dict[str, Any] = {
|
|
"schema": "crewai.flow/v1",
|
|
"name": "DeclaredChat",
|
|
"state": {
|
|
"type": "pydantic",
|
|
"ref": "crewai.experimental.conversational:ConversationState",
|
|
},
|
|
"conversational": {},
|
|
"methods": {
|
|
"route_conversation": {
|
|
"do": {"call": "code", "ref": f"{_MIXIN}.route_conversation"},
|
|
"start": True,
|
|
"router": True,
|
|
},
|
|
"converse_turn": {
|
|
"do": {"call": "code", "ref": f"{_MIXIN}.converse_turn"},
|
|
"listen": "converse",
|
|
},
|
|
"end_conversation": {
|
|
"do": {"call": "code", "ref": f"{_MIXIN}.end_conversation"},
|
|
"listen": "end",
|
|
},
|
|
},
|
|
}
|
|
declaration.update(overrides)
|
|
return declaration
|
|
|
|
|
|
class TestDeclarativeConversationalFlow:
|
|
"""A declaration's ``conversational`` block drives the runtime."""
|
|
|
|
def test_declaration_enables_conversational_mode(self) -> None:
|
|
flow = Flow.from_declaration(contents=_conversational_declaration())
|
|
|
|
assert flow._is_conversational_enabled() is True
|
|
|
|
def test_declaration_with_enabled_false_stays_non_conversational(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents=_conversational_declaration(conversational={"enabled": False})
|
|
)
|
|
|
|
assert flow._is_conversational_enabled() is False
|
|
|
|
def test_declared_route_labels_reach_the_router_catalog(self) -> None:
|
|
declaration = _conversational_declaration()
|
|
declaration["methods"]["handle_order"] = {
|
|
"do": {"call": "expression", "expr": "'shipped'"},
|
|
"listen": "order",
|
|
"description": "Order status questions.",
|
|
}
|
|
|
|
flow = Flow.from_declaration(contents=declaration)
|
|
|
|
assert "order" in flow._valid_route_labels()
|
|
assert "order" in flow._effective_routes()
|
|
|
|
def test_declaration_runs_a_turn_and_accumulates_history(self) -> None:
|
|
declaration = _conversational_declaration(
|
|
conversational={"system_prompt": "You are terse."}
|
|
)
|
|
chat = _ScriptedLLM(["Hi there.", "Anything else?"])
|
|
flow = Flow.from_declaration(contents=declaration)
|
|
flow._conversation_config.llm = chat
|
|
|
|
assert flow.handle_turn("hello") == "Hi there."
|
|
assert flow.handle_turn("thanks") == "Anything else?"
|
|
assert [(m.role, m.content) for m in flow.state.messages] == [
|
|
("user", "hello"),
|
|
("assistant", "Hi there."),
|
|
("user", "thanks"),
|
|
("assistant", "Anything else?"),
|
|
]
|
|
assert chat.seen[-1][0] == {"role": "system", "content": "You are terse."}
|
|
|
|
def test_declaration_drives_deferred_trace_finalization(self) -> None:
|
|
deferred = Flow.from_declaration(contents=_conversational_declaration())
|
|
not_deferred = Flow.from_declaration(
|
|
contents=_conversational_declaration(
|
|
conversational={"defer_trace_finalization": False}
|
|
)
|
|
)
|
|
|
|
assert deferred._should_defer_trace_finalization() is True
|
|
assert not_deferred._should_defer_trace_finalization() is False
|
|
|
|
def test_declaration_orders_the_router_last_and_sequentially(self) -> None:
|
|
flow = Flow.from_declaration(contents=_conversational_declaration())
|
|
|
|
ordered, sequential = flow._order_start_methods_for_kickoff(
|
|
["bootstrap", "route_conversation"]
|
|
)
|
|
|
|
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"
|
|
)
|
|
flow = Flow.from_declaration(
|
|
contents=_conversational_declaration(
|
|
conversational={
|
|
"router": {"response_format": {"ref": "some.module:Schema"}}
|
|
}
|
|
)
|
|
)
|
|
|
|
assert flow._conversation_config.router.response_format is None
|
|
assert "cannot carry a model class" in caplog.text
|
|
|
|
|
|
class DeclaredSchemaChatState(ConversationState):
|
|
"""A ConversationState subclass, i.e. the already-supported state shape."""
|
|
|
|
ticket_id: str | None = None
|
|
|
|
|
|
class TestDeclaredConversationalState:
|
|
"""Any declared state shape works for a chat flow, keeping its own fields."""
|
|
|
|
@staticmethod
|
|
def _flow(state: dict[str, Any] | None) -> Flow[Any]:
|
|
declaration = _conversational_declaration(conversational={})
|
|
if state is None:
|
|
declaration.pop("state", None)
|
|
else:
|
|
declaration["state"] = state
|
|
return Flow.from_declaration(contents=declaration)
|
|
|
|
CONVERSATIONAL_FIELDS = (
|
|
"messages",
|
|
"current_user_message",
|
|
"last_intent",
|
|
"ended",
|
|
"events",
|
|
"agent_threads",
|
|
)
|
|
|
|
def test_inline_json_schema_state_gains_the_conversational_fields(self) -> None:
|
|
flow = self._flow(
|
|
{
|
|
"type": "json_schema",
|
|
"json_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"ticket_id": {"type": "string"},
|
|
"turns": {"type": "integer"},
|
|
},
|
|
},
|
|
"default": {"ticket_id": "T-1"},
|
|
}
|
|
)
|
|
fields = type(flow.state).model_fields
|
|
|
|
for name in self.CONVERSATIONAL_FIELDS:
|
|
assert name in fields, name
|
|
assert "ticket_id" in fields and "turns" in fields
|
|
assert flow.state.ticket_id == "T-1"
|
|
assert flow.state.id
|
|
|
|
def test_a_pydantic_ref_that_is_not_a_conversation_state_is_composed(self) -> None:
|
|
flow = self._flow({"type": "pydantic", "ref": "crewai.flow:ChatState"})
|
|
fields = type(flow.state).model_fields
|
|
|
|
for name in self.CONVERSATIONAL_FIELDS:
|
|
assert name in fields, name
|
|
assert "session_ready" in fields
|
|
|
|
def test_a_conversation_state_subclass_is_not_composed_twice(self) -> None:
|
|
flow = self._flow(
|
|
{"type": "pydantic", "ref": f"{__name__}:DeclaredSchemaChatState"}
|
|
)
|
|
mro = [cls.__name__ for cls in type(flow.state).__mro__]
|
|
|
|
assert mro.count("ConversationState") == 1
|
|
assert "ticket_id" in type(flow.state).model_fields
|
|
|
|
def test_dict_state_is_given_the_real_shape(self) -> None:
|
|
"""A dict cannot carry the fields, so supply them rather than die."""
|
|
flow = self._flow({"type": "dict", "default": {"last_intent": "order"}})
|
|
|
|
assert isinstance(flow.state, ConversationState)
|
|
assert flow.state.last_intent == "order"
|
|
|
|
def test_dict_state_keeps_declared_defaults_it_does_not_own(self) -> None:
|
|
"""Dropping them would break an action reading `state.topic`."""
|
|
flow = self._flow({"type": "dict", "default": {"topic": "ai", "limit": 3}})
|
|
|
|
assert flow.state.topic == "ai"
|
|
assert flow.state.limit == 3
|
|
assert isinstance(flow.state, ConversationState)
|
|
|
|
def test_unknown_state_is_treated_like_dict(self) -> None:
|
|
flow = self._flow({"type": "unknown", "ref": "x:Y", "default": {"topic": "ai"}})
|
|
|
|
assert flow.state.topic == "ai"
|
|
assert isinstance(flow.state, ConversationState)
|
|
|
|
def test_an_unbuildable_declared_model_still_gets_the_chat_shape(self) -> None:
|
|
"""A bad ref fell back to a plain dict, so the turn died on `state.id`."""
|
|
flow = self._flow({"type": "pydantic", "ref": "no.such.module:Nope"})
|
|
|
|
assert isinstance(flow.state, ConversationState)
|
|
assert flow.state.id
|
|
|
|
def test_declared_state_still_runs_a_turn(self) -> None:
|
|
flow = self._flow(
|
|
{
|
|
"type": "json_schema",
|
|
"json_schema": {
|
|
"type": "object",
|
|
"properties": {"ticket_id": {"type": "string"}},
|
|
},
|
|
}
|
|
)
|
|
flow._conversation_config.llm = _ScriptedLLM(["Hello."])
|
|
|
|
assert flow.handle_turn("hi") == "Hello."
|
|
assert [m.role for m in flow.state.messages] == ["user", "assistant"]
|
|
|
|
def test_non_conversational_declaration_keeps_its_plain_state(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents={
|
|
"schema": "crewai.flow/v1",
|
|
"name": "Plain",
|
|
"state": {
|
|
"type": "json_schema",
|
|
"json_schema": {
|
|
"type": "object",
|
|
"properties": {"topic": {"type": "string"}},
|
|
},
|
|
"default": {"topic": "ai"},
|
|
},
|
|
"methods": {
|
|
"begin": {
|
|
"do": {"call": "expression", "expr": "state.topic"},
|
|
"start": True,
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert "messages" not in type(flow.state).model_fields
|
|
assert flow.state.topic == "ai"
|
|
|
|
class TestDeclaredConversationalLLM:
|
|
"""A conversational block accepts the shapes a crew agent's `llm` accepts."""
|
|
|
|
@staticmethod
|
|
def _resolved(llm: Any) -> Any:
|
|
flow = Flow.from_declaration(
|
|
contents=_conversational_declaration(conversational={"llm": llm})
|
|
)
|
|
return flow._coerce_llm(flow._conversation_config.llm)
|
|
|
|
def test_model_id_string(self) -> None:
|
|
assert self._resolved("gpt-4o-mini").model == "gpt-4o-mini"
|
|
|
|
def test_config_mapping_carries_provider_settings(self) -> None:
|
|
resolved = self._resolved({"model": "openai/gpt-4o-mini", "max_tokens": 512})
|
|
|
|
assert resolved.model == "gpt-4o-mini"
|
|
assert resolved.max_tokens == 512
|
|
|
|
def test_config_mapping_passes_through_extra_settings(self) -> None:
|
|
resolved = self._resolved({"model": "openai/gpt-4o-mini", "temperature": 0.2})
|
|
|
|
assert resolved.temperature == 0.2
|
|
|
|
def test_a_live_llm_object_is_passed_through_untouched(self) -> None:
|
|
llm = _ScriptedLLM(["hi"])
|
|
|
|
@ConversationConfig(llm=llm)
|
|
class ClassChat(Flow[ConversationState]):
|
|
pass
|
|
|
|
assert ClassChat()._coerce_llm(llm) is llm
|
|
|
|
def test_a_mapping_without_a_model_key_says_so(self) -> None:
|
|
with pytest.raises(ValueError, match="must include 'model'"):
|
|
self._resolved({"max_tokens": 512})
|
|
|
|
def test_a_wrong_type_fails_now_not_at_call_time(self) -> None:
|
|
"""`create_llm` would take an int as a model name and fail later."""
|
|
with pytest.raises(ValueError, match="expected a model-id string"):
|
|
self._resolved(1234)
|
|
|
|
def test_an_llm_definition_resolves_with_its_settings(self) -> None:
|
|
from crewai.project.crew_definition import LLMDefinition
|
|
|
|
flow = Flow.from_declaration(contents=_conversational_declaration())
|
|
resolved = flow._coerce_llm(
|
|
LLMDefinition(model="openai/gpt-4o-mini", max_tokens=256)
|
|
)
|
|
|
|
assert resolved.model == "gpt-4o-mini"
|
|
assert resolved.max_tokens == 256
|
|
|
|
def test_a_declared_mapping_reaches_create_llm_during_a_turn(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Swapping the config out before the turn would not prove the path."""
|
|
declared = {"model": "openai/gpt-4o-mini", "max_tokens": 128}
|
|
scripted = _ScriptedLLM(["Hello."])
|
|
seen: list[Any] = []
|
|
|
|
def fake_create_llm(value: Any) -> Any:
|
|
seen.append(value)
|
|
return scripted
|
|
|
|
monkeypatch.setattr(
|
|
"crewai.utilities.llm_utils.create_llm", fake_create_llm
|
|
)
|
|
flow = Flow.from_declaration(
|
|
contents=_conversational_declaration(conversational={"llm": declared})
|
|
)
|
|
|
|
assert flow.handle_turn("hi") == "Hello."
|
|
assert declared in seen
|
|
|
|
def test_a_declared_intent_llm_mapping_is_resolved(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""`_collapse_to_outcome` takes only str | BaseLLM, so it must be coerced.
|
|
|
|
Swapping the mapping for an LLM before the turn would skip the coercion
|
|
entirely, so the mapping stays declared and ``create_llm`` is patched.
|
|
"""
|
|
declared = {"model": "openai/gpt-4o-mini", "max_tokens": 64}
|
|
intent_llm = _ScriptedLLM(['{"outcome": "order"}'])
|
|
converse_llm = _ScriptedLLM(["ok"])
|
|
seen: list[Any] = []
|
|
|
|
def fake_create_llm(value: Any) -> Any:
|
|
seen.append(value)
|
|
return intent_llm if value == declared else converse_llm
|
|
|
|
monkeypatch.setattr("crewai.utilities.llm_utils.create_llm", fake_create_llm)
|
|
flow = Flow.from_declaration(
|
|
contents=_conversational_declaration(
|
|
conversational={"default_intents": ["order"], "intent_llm": declared}
|
|
)
|
|
)
|
|
|
|
# Before coercion this raised "Invalid llm type: <class 'dict'>".
|
|
flow.handle_turn("where is my order?")
|
|
|
|
assert declared in seen
|
|
assert [m.role for m in flow.state.messages][0] == "user"
|
|
|
|
class TestRoutingArtefactLabels:
|
|
"""A route label echoed by a handler is a routing artefact, not a reply."""
|
|
|
|
def test_class_based_set_matches_the_framework_labels(self) -> None:
|
|
@ConversationConfig()
|
|
class ClassChat(Flow[ConversationState]):
|
|
pass
|
|
|
|
assert ClassChat()._routing_artefact_labels() == {
|
|
"conversation",
|
|
"converse",
|
|
"end",
|
|
"answer_from_history",
|
|
"route_to_flow",
|
|
}
|
|
|
|
def test_a_declared_extra_builtin_route_is_covered(self) -> None:
|
|
"""A literal list would miss it; deriving from the routes does not."""
|
|
flow = Flow.from_declaration(
|
|
contents=_conversational_declaration(
|
|
conversational={"builtin_routes": ["converse", "end", "greet"]}
|
|
)
|
|
)
|
|
|
|
assert "greet" in flow._routing_artefact_labels()
|
|
assert flow._is_public_turn_result("greet") is False
|
|
|
|
def test_ordinary_text_is_still_a_reply(self) -> None:
|
|
flow = Flow.from_declaration(contents=_conversational_declaration())
|
|
|
|
assert flow._is_public_turn_result("Your order shipped.") is True
|
|
|
|
|
|
class TestConversationalCapabilityAttribute:
|
|
"""A declarative chat flow reports itself conversational to outside callers.
|
|
|
|
Consumers outside this package capability-check the ``conversational``
|
|
attribute, so it must agree with ``_is_conversational_enabled()``.
|
|
"""
|
|
|
|
def test_declaration_marks_the_instance_conversational(self) -> None:
|
|
flow = Flow.from_declaration(contents=_conversational_declaration())
|
|
|
|
assert getattr(flow, "conversational") is True
|
|
assert flow._is_conversational_enabled() is True
|
|
assert callable(getattr(flow, "stream_turn", None))
|
|
|
|
def test_the_flag_does_not_leak_onto_the_class(self) -> None:
|
|
"""The DSL projection reads it off the class to decide what to emit."""
|
|
Flow.from_declaration(contents=_conversational_declaration())
|
|
|
|
assert Flow.conversational is False
|
|
|
|
class LaterPlainFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "begin"
|
|
|
|
assert LaterPlainFlow.conversational is False
|
|
assert LaterPlainFlow.flow_definition().conversational is None
|
|
|
|
def test_disabled_declaration_is_not_marked(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents=_conversational_declaration(conversational={"enabled": False})
|
|
)
|
|
|
|
assert getattr(flow, "conversational") is False
|
|
|
|
def test_non_conversational_declaration_is_not_marked(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents={
|
|
"schema": "crewai.flow/v1",
|
|
"name": "Plain",
|
|
"methods": {
|
|
"begin": {
|
|
"do": {"call": "expression", "expr": "'x'"},
|
|
"start": True,
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert getattr(flow, "conversational") is False
|
|
|
|
def test_class_based_flow_keeps_its_own_flag(self) -> None:
|
|
@ConversationConfig()
|
|
class ClassChat(Flow[ConversationState]):
|
|
pass
|
|
|
|
assert ClassChat.conversational is True
|
|
assert getattr(ClassChat(), "conversational") is True
|
|
|
|
|
|
class TestBuiltinRouteResolution:
|
|
"""``route_turn`` and ``_effective_routes`` agree on what is built in."""
|
|
|
|
@staticmethod
|
|
def _flow(builtin: list[str]) -> Flow[Any]:
|
|
return Flow.from_declaration(
|
|
contents=_conversational_declaration(
|
|
conversational={"builtin_routes": builtin}
|
|
)
|
|
)
|
|
|
|
def test_declared_builtin_routes_are_not_mistaken_for_custom_ones(self) -> None:
|
|
"""A declaration adding a builtin route must not auto-enable the router.
|
|
|
|
``route_turn`` used to subtract the class attribute while
|
|
``_effective_routes`` used the declaration, so an added builtin looked
|
|
like a custom route and turned the LLM router on for every turn.
|
|
"""
|
|
flow = self._flow(["converse", "end", "greet"])
|
|
|
|
assert flow._effective_builtin_routes() == {"converse", "end", "greet"}
|
|
assert flow.route_turn(flow.build_router_context()) is None
|
|
|
|
def test_a_real_custom_route_still_auto_enables_the_router(self) -> None:
|
|
declaration = _conversational_declaration(conversational={})
|
|
declaration["methods"]["handle_order"] = {
|
|
"do": {"call": "expression", "expr": "'shipped'"},
|
|
"listen": "order",
|
|
}
|
|
flow = Flow.from_declaration(contents=declaration)
|
|
flow._conversation_config.llm = _ScriptedLLM(['{"intent": "order"}'])
|
|
|
|
assert "order" in flow._effective_routes(None) - flow._effective_builtin_routes()
|
|
assert flow.route_turn(flow.build_router_context()) == "order"
|
|
|
|
def test_class_based_flow_falls_back_to_its_class_attributes(self) -> None:
|
|
@ConversationConfig()
|
|
class ClassChat(Flow[ConversationState]):
|
|
pass
|
|
|
|
flow = ClassChat()
|
|
|
|
assert flow._effective_builtin_routes() == set(flow.builtin_routes)
|
|
assert flow._effective_internal_routes() == set(flow.internal_routes)
|
|
|
|
|
|
class TestConversationalStatePrecedence:
|
|
"""A declared ``state:`` block is never replaced by the default."""
|
|
|
|
def test_declared_state_survives_on_both_paths(self) -> None:
|
|
class SubclassChat(Flow):
|
|
conversational = True
|
|
|
|
declaration = _conversational_declaration(
|
|
state={"type": "pydantic", "ref": f"{__name__}:DeclaredChatState"}
|
|
)
|
|
|
|
for cls in (Flow, SubclassChat):
|
|
state = cls.from_declaration(contents=declaration).state
|
|
|
|
assert "ticket_id" in type(state).model_fields, cls.__name__
|
|
assert "messages" in type(state).model_fields, cls.__name__
|
|
|
|
def test_conversation_state_is_implied_when_none_declared(self) -> None:
|
|
declaration = _conversational_declaration()
|
|
del declaration["state"]
|
|
|
|
flow = Flow.from_declaration(contents=declaration)
|
|
|
|
assert isinstance(flow.state, ConversationState)
|
|
|
|
def test_non_conversational_declaration_keeps_its_state(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents={
|
|
"schema": "crewai.flow/v1",
|
|
"name": "Plain",
|
|
"state": {"type": "dict", "default": {"topic": "ai"}},
|
|
"methods": {
|
|
"begin": {
|
|
"do": {"call": "expression", "expr": "state.topic"},
|
|
"start": True,
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert flow.state["topic"] == "ai"
|
|
|
|
|
|
class TestClassConfigStillWins:
|
|
"""Existing decorated Python flows keep their live objects."""
|
|
|
|
def test_live_llm_object_is_not_downgraded_by_the_definition(self) -> None:
|
|
llm = _ScriptedLLM(["from the live object"])
|
|
|
|
@ConversationConfig(llm=llm)
|
|
class LiveLLMFlow(ConversationalFlow):
|
|
pass
|
|
|
|
flow = LiveLLMFlow()
|
|
|
|
assert flow._conversation_config.llm is llm
|
|
assert flow.handle_turn("hello") == "from the live object"
|
|
|
|
def test_live_router_response_format_is_not_downgraded(self) -> None:
|
|
class MyRoute(BaseModel):
|
|
intent: str
|
|
|
|
@ConversationConfig(router=RouterConfig(response_format=MyRoute))
|
|
class LiveFormatFlow(ConversationalFlow):
|
|
pass
|
|
|
|
assert LiveFormatFlow()._conversation_config.router.response_format is MyRoute
|
|
|
|
def test_defer_trace_finalization_follows_the_class_config(self) -> None:
|
|
"""Deferral is a behavior knob, so it follows the same precedence.
|
|
|
|
The class config and the declaration can disagree on the hybrid path;
|
|
every other setting follows the class config, and this must too.
|
|
"""
|
|
|
|
@ConversationConfig(defer_trace_finalization=False)
|
|
class ConfiguredChat(Flow):
|
|
conversational = True
|
|
|
|
flow = ConfiguredChat.from_declaration(
|
|
contents=_conversational_declaration(conversational={})
|
|
)
|
|
|
|
assert flow._conversation_definition.defer_trace_finalization is True
|
|
assert flow._conversation_config.defer_trace_finalization is False
|
|
assert flow._should_defer_trace_finalization() is False
|
|
|
|
def test_instance_flag_still_forces_deferral(self) -> None:
|
|
@ConversationConfig(defer_trace_finalization=False)
|
|
class ConfiguredChat(Flow):
|
|
conversational = True
|
|
|
|
flow = ConfiguredChat()
|
|
assert flow._should_defer_trace_finalization() is False
|
|
|
|
flow.defer_trace_finalization = True
|
|
assert flow._should_defer_trace_finalization() is True
|
|
|
|
def test_non_conversational_flow_never_defers_from_config(self) -> None:
|
|
class PlainFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "begin"
|
|
|
|
assert PlainFlow()._should_defer_trace_finalization() is False
|
|
|
|
def test_class_config_wins_over_a_declaration_block(self) -> None:
|
|
llm = _ScriptedLLM(["from the class config"])
|
|
|
|
@ConversationConfig(llm=llm, system_prompt="From the class.")
|
|
class ConfiguredChat(Flow):
|
|
conversational = True
|
|
|
|
flow = ConfiguredChat.from_declaration(
|
|
contents=_conversational_declaration(
|
|
conversational={"system_prompt": "From the declaration."}
|
|
)
|
|
)
|
|
|
|
assert flow._conversation_config.llm is llm
|
|
assert flow._conversation_config.system_prompt == "From the class."
|
|
|
|
|
|
class TestMinimalDeclarativeChat:
|
|
"""A declaration only has to name its own routes."""
|
|
|
|
@staticmethod
|
|
def _declaration(**overrides: Any) -> dict[str, Any]:
|
|
declaration: dict[str, Any] = {
|
|
"schema": "crewai.flow/v1",
|
|
"name": "MinimalChat",
|
|
"conversational": {},
|
|
"methods": {
|
|
"handle_order": {
|
|
"do": {"call": "expression", "expr": "'Your order shipped.'"},
|
|
"listen": "order",
|
|
"description": "Order status questions.",
|
|
}
|
|
},
|
|
}
|
|
declaration.update(overrides)
|
|
return declaration
|
|
|
|
def test_minimal_declaration_runs_a_conversational_turn(self) -> None:
|
|
# No custom routes, so the router does not auto-enable and the turn
|
|
# goes straight to the built-in converse handler.
|
|
declaration = self._declaration(methods={})
|
|
flow = Flow.from_declaration(contents=declaration)
|
|
flow._conversation_config.llm = _ScriptedLLM(["Hello there."])
|
|
|
|
assert flow.handle_turn("hi") == "Hello there."
|
|
assert [(m.role, m.content) for m in flow.state.messages] == [
|
|
("user", "hi"),
|
|
("assistant", "Hello there."),
|
|
]
|
|
|
|
def test_minimal_declaration_implies_conversation_state(self) -> None:
|
|
flow = Flow.from_declaration(contents=self._declaration())
|
|
|
|
assert isinstance(flow.state, ConversationState)
|
|
|
|
def test_minimal_declaration_routes_to_a_declared_handler(self) -> None:
|
|
flow = Flow.from_declaration(contents=self._declaration())
|
|
flow._conversation_config.router = RouterConfig(
|
|
llm=_ScriptedLLM(['{"intent": "order"}'])
|
|
)
|
|
|
|
assert flow.handle_turn("where is my order?") == "Your order shipped."
|
|
assert flow.state.last_intent == "order"
|
|
|
|
def test_method_description_reaches_the_router_catalog(self) -> None:
|
|
flow = Flow.from_declaration(contents=self._declaration())
|
|
|
|
catalog = flow._build_route_catalog(RouterConfig(routes=["order"]))
|
|
|
|
assert catalog["order"] == "Order status questions."
|
|
|
|
|
|
class TestBuiltinMethodSynthesis:
|
|
"""A declaration does not have to name the built-in methods."""
|
|
|
|
BUILTIN_METHODS = {
|
|
"route_conversation",
|
|
"converse_turn",
|
|
"end_conversation",
|
|
"answer_from_history_turn",
|
|
}
|
|
|
|
@staticmethod
|
|
def _declaration(**overrides: Any) -> dict[str, Any]:
|
|
declaration: dict[str, Any] = {
|
|
"schema": "crewai.flow/v1",
|
|
"name": "SynthChat",
|
|
"conversational": {},
|
|
"methods": {
|
|
"handle_order": {
|
|
"do": {"call": "expression", "expr": "'shipped'"},
|
|
"listen": "order",
|
|
}
|
|
},
|
|
}
|
|
declaration.update(overrides)
|
|
return declaration
|
|
|
|
def test_builtin_methods_are_synthesized_with_their_roles(self) -> None:
|
|
flow = Flow.from_declaration(contents=self._declaration())
|
|
methods = flow._definition.methods
|
|
|
|
assert self.BUILTIN_METHODS <= set(methods)
|
|
assert methods["route_conversation"].start is True
|
|
assert methods["route_conversation"].router is True
|
|
assert methods["converse_turn"].listen == "converse"
|
|
assert methods["end_conversation"].listen == "end"
|
|
assert methods["answer_from_history_turn"].listen == "answer_from_history"
|
|
|
|
def test_synthesized_refs_match_the_python_projection(self) -> None:
|
|
class ProjectedChat(Flow):
|
|
conversational = True
|
|
|
|
projected = ProjectedChat.flow_definition().methods
|
|
synthesized = Flow.from_declaration(
|
|
contents=self._declaration()
|
|
)._definition.methods
|
|
|
|
for name in self.BUILTIN_METHODS:
|
|
assert synthesized[name].do == projected[name].do
|
|
|
|
def test_author_supplied_entry_is_not_overridden(self) -> None:
|
|
declaration = self._declaration()
|
|
declaration["methods"]["converse_turn"] = {
|
|
"do": {"call": "expression", "expr": "'mine'"},
|
|
"listen": "converse",
|
|
}
|
|
|
|
flow = Flow.from_declaration(contents=declaration)
|
|
|
|
assert flow._definition.methods["converse_turn"].do.expr == "'mine'"
|
|
|
|
def test_disabled_block_synthesizes_nothing(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents=self._declaration(conversational={"enabled": False})
|
|
)
|
|
|
|
assert set(flow._definition.methods) == {"handle_order"}
|
|
|
|
def test_non_conversational_flow_synthesizes_nothing(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents={
|
|
"schema": "crewai.flow/v1",
|
|
"name": "Plain",
|
|
"methods": {
|
|
"begin": {
|
|
"do": {"call": "expression", "expr": "'x'"},
|
|
"start": True,
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert set(flow._definition.methods) == {"begin"}
|
|
|
|
def test_the_loaded_definition_itself_is_left_alone(self) -> None:
|
|
"""Synthesis is a runtime concern; the contract stays as authored."""
|
|
from crewai.flow.flow_definition import FlowDefinition
|
|
|
|
definition = FlowDefinition.from_declaration(contents=self._declaration())
|
|
|
|
assert set(definition.methods) == {"handle_order"}
|
|
|
|
|
|
class TestRouteDescriptions:
|
|
"""Route descriptions survive into a declaration."""
|
|
|
|
def test_python_handler_docstring_is_projected(self) -> None:
|
|
class DocumentedChat(Flow):
|
|
conversational = True
|
|
|
|
@listen("research")
|
|
def handle_research(self) -> str:
|
|
"""Fresh web research and current news."""
|
|
return "researched"
|
|
|
|
definition = DocumentedChat.flow_definition()
|
|
|
|
assert (
|
|
definition.methods["handle_research"].description
|
|
== "Fresh web research and current news."
|
|
)
|
|
|
|
def test_declared_description_reaches_the_catalog(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents={
|
|
"schema": "crewai.flow/v1",
|
|
"name": "DescribedChat",
|
|
"conversational": {},
|
|
"methods": {
|
|
"handle_order": {
|
|
"do": {"call": "expression", "expr": "'shipped'"},
|
|
"listen": "order",
|
|
"description": "Order status questions.",
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
catalog = flow._build_route_catalog(RouterConfig(routes=["order"]))
|
|
|
|
assert catalog["order"] == "Order status questions."
|
|
|
|
def test_missing_description_is_empty_not_nonetype_docstring(self) -> None:
|
|
flow = Flow.from_declaration(
|
|
contents={
|
|
"schema": "crewai.flow/v1",
|
|
"name": "UndescribedChat",
|
|
"conversational": {},
|
|
"methods": {
|
|
"handle_order": {
|
|
"do": {"call": "expression", "expr": "'shipped'"},
|
|
"listen": "order",
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
catalog = flow._build_route_catalog(RouterConfig(routes=["order"]))
|
|
|
|
assert catalog["order"] == ""
|
|
|
|
def test_python_docstring_still_describes_the_route(self) -> None:
|
|
class DocumentedChat(Flow):
|
|
conversational = True
|
|
|
|
@listen("research")
|
|
def handle_research(self) -> str:
|
|
"""Fresh web research and current news."""
|
|
return "researched"
|
|
|
|
catalog = DocumentedChat()._build_route_catalog(
|
|
RouterConfig(routes=["research"])
|
|
)
|
|
|
|
assert catalog["research"] == "Fresh web research and current news."
|
|
|
|
|
|
class TestDeclarativeTurnMatrix:
|
|
"""A declaration-built flow across the turn entry points."""
|
|
|
|
@staticmethod
|
|
def _flow(reply: str = "Declared reply.") -> Flow[Any]:
|
|
flow = Flow.from_declaration(
|
|
contents={
|
|
"schema": "crewai.flow/v1",
|
|
"name": "MatrixChat",
|
|
"conversational": {},
|
|
"methods": {},
|
|
}
|
|
)
|
|
flow._conversation_config.llm = _ScriptedLLM([reply])
|
|
return flow
|
|
|
|
def test_sync_turn(self) -> None:
|
|
assert self._flow().handle_turn("hi") == "Declared reply."
|
|
|
|
def test_stream_turn_frames_match_the_class_based_path(self) -> None:
|
|
"""A chat UI must see the same frame sequence either way."""
|
|
|
|
@ConversationConfig(llm=_ScriptedLLM(["streamed reply"]))
|
|
class ClassChat(Flow[ConversationState]):
|
|
pass
|
|
|
|
class_frames = [f.type for f in ClassChat().stream_turn("hi").events]
|
|
|
|
flow = self._flow("streamed reply")
|
|
stream = flow.stream_turn("hi", session_id="session-1")
|
|
declared_frames = [f.type for f in stream.events]
|
|
|
|
assert declared_frames == class_frames
|
|
assert declared_frames[0] == "conversation_turn_started"
|
|
assert declared_frames[-1] == "conversation_turn_completed"
|
|
assert "conversation_message_added" in declared_frames
|
|
assert stream.result == "streamed reply"
|
|
assert flow.state.messages[-1].content == "streamed reply"
|
|
|
|
def test_turn_inside_a_running_event_loop(self) -> None:
|
|
"""``kickoff`` takes its thread-pool path when a loop is already running."""
|
|
import asyncio
|
|
|
|
flow = self._flow("reply from the loop")
|
|
|
|
async def run() -> Any:
|
|
return flow.handle_turn("hi")
|
|
|
|
assert asyncio.run(run()) == "reply from the loop"
|
|
assert flow.state.messages[-1].content == "reply from the loop"
|
|
|
|
def test_follow_up_turn_keeps_history(self) -> None:
|
|
flow = self._flow()
|
|
flow._conversation_config.llm = _ScriptedLLM(["first", "second"])
|
|
|
|
flow.handle_turn("one")
|
|
flow.handle_turn("two")
|
|
|
|
assert [(m.role, m.content) for m in flow.state.messages] == [
|
|
("user", "one"),
|
|
("assistant", "first"),
|
|
("user", "two"),
|
|
("assistant", "second"),
|
|
]
|
|
|
|
def test_chat_repl_drives_declared_turns(self) -> None:
|
|
flow = self._flow()
|
|
flow._conversation_config.llm = _ScriptedLLM(["hello", "goodbye"])
|
|
prompts: list[str] = []
|
|
outputs: list[str] = []
|
|
|
|
replies = iter(["hi", "bye", "exit"])
|
|
|
|
def input_fn(prompt: str) -> str:
|
|
prompts.append(prompt)
|
|
return next(replies)
|
|
|
|
flow.chat(input_fn=input_fn, output_fn=outputs.append)
|
|
|
|
assert [m.content for m in flow.state.messages if m.role == "user"] == [
|
|
"hi",
|
|
"bye",
|
|
]
|
|
assert len(outputs) == 2
|
|
|
|
|
|
class TestPrivateAgentResultsStayPrivate:
|
|
"""Unwrapping ``.raw`` must not defeat ``visible_agent_outputs``."""
|
|
|
|
class _Out:
|
|
"""Shape of ``LiteAgentOutput`` / ``CrewOutput``."""
|
|
|
|
def __init__(self, raw: str) -> None:
|
|
self.raw = raw
|
|
|
|
def test_privately_recorded_result_is_not_republished(self) -> None:
|
|
"""A handler that records privately and hands the object back.
|
|
|
|
``append_agent_result`` defaults to private and does not set the
|
|
reply flag, so before this guard the end-of-turn fallback promoted the
|
|
very object the handler had just asked to keep out of the transcript.
|
|
"""
|
|
recorded = self._Out("SECRET scratch work")
|
|
|
|
@ConversationConfig()
|
|
class PrivateChat(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> Any:
|
|
self.append_agent_result("researcher", recorded)
|
|
return recorded
|
|
|
|
flow = PrivateChat()
|
|
flow.handle_turn("look something up")
|
|
|
|
assert [m.role for m in flow.state.messages] == ["user"]
|
|
assert flow.state.agent_threads["researcher"][0].content == (
|
|
"SECRET scratch work"
|
|
)
|
|
|
|
def test_a_summary_returned_alongside_a_private_result_is_published(self) -> None:
|
|
"""Identity, not content: a different return value is still promoted."""
|
|
|
|
@ConversationConfig()
|
|
class SummarisingChat(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> Any:
|
|
self.append_agent_result(
|
|
"researcher", TestPrivateAgentResultsStayPrivate._Out("SECRET")
|
|
)
|
|
return "Here is the summary."
|
|
|
|
flow = SummarisingChat()
|
|
flow.handle_turn("look something up")
|
|
|
|
assert flow.state.messages[-1].content == "Here is the summary."
|
|
|
|
def test_visible_agent_outputs_still_publishes(self) -> None:
|
|
result = self._Out("PUBLIC RESULT")
|
|
|
|
@ConversationConfig(visible_agent_outputs="all")
|
|
class VisibleChat(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> Any:
|
|
self.append_agent_result("researcher", result)
|
|
return result
|
|
|
|
flow = VisibleChat()
|
|
flow.handle_turn("go")
|
|
|
|
assert flow.state.messages[-1].content == "PUBLIC RESULT"
|
|
assert [m.role for m in flow.state.messages].count("assistant") == 1
|
|
|
|
def test_explicit_public_visibility_still_publishes(self) -> None:
|
|
result = self._Out("PUBLIC RESULT")
|
|
|
|
@ConversationConfig()
|
|
class PublicChat(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> Any:
|
|
self.append_agent_result("researcher", result, visibility="public")
|
|
return result
|
|
|
|
flow = PublicChat()
|
|
flow.handle_turn("go")
|
|
|
|
assert flow.state.messages[-1].content == "PUBLIC RESULT"
|
|
assert [m.role for m in flow.state.messages].count("assistant") == 1
|
|
|
|
def test_recorded_results_do_not_leak_across_turns(self) -> None:
|
|
"""The per-turn list resets, so turn 2 is judged on its own."""
|
|
shared = self._Out("reply text")
|
|
|
|
@ConversationConfig()
|
|
class TwoTurnChat(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> Any:
|
|
return shared
|
|
|
|
flow = TwoTurnChat()
|
|
flow.handle_turn("one")
|
|
flow.handle_turn("two")
|
|
|
|
assert [m.content for m in flow.state.messages if m.role == "assistant"] == [
|
|
"reply text",
|
|
"reply text",
|
|
]
|
|
|
|
|
|
class TestHandlerReplyPromotion:
|
|
"""An agent or crew handler's reply reaches the transcript."""
|
|
|
|
class _AgentOutput:
|
|
"""Shape of ``LiteAgentOutput`` / ``CrewOutput``: text lives on ``.raw``."""
|
|
|
|
def __init__(self, raw: str) -> None:
|
|
self.raw = raw
|
|
|
|
def __str__(self) -> str:
|
|
return f"<output {self.raw!r}>"
|
|
|
|
@staticmethod
|
|
def _chat(handler_result: Any) -> Flow[Any]:
|
|
@ConversationConfig()
|
|
class HandlerChat(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> Any:
|
|
return handler_result
|
|
|
|
return HandlerChat()
|
|
|
|
def test_output_object_reply_reaches_history(self) -> None:
|
|
flow = self._chat(self._AgentOutput("Your order shipped Tuesday."))
|
|
|
|
flow.handle_turn("where is my order?")
|
|
|
|
assert [(m.role, m.content) for m in flow.state.messages] == [
|
|
("user", "where is my order?"),
|
|
("assistant", "Your order shipped Tuesday."),
|
|
]
|
|
|
|
def test_string_reply_still_reaches_history(self) -> None:
|
|
flow = self._chat("A plain string reply.")
|
|
|
|
flow.handle_turn("hi")
|
|
|
|
assert flow.state.messages[-1].content == "A plain string reply."
|
|
|
|
def test_route_label_output_is_not_promoted(self) -> None:
|
|
flow = self._chat(self._AgentOutput("converse"))
|
|
|
|
flow.handle_turn("hi")
|
|
|
|
assert [m.role for m in flow.state.messages] == ["user"]
|
|
|
|
def test_output_matching_this_turns_intent_is_not_promoted(self) -> None:
|
|
@ConversationConfig()
|
|
class RoutedChat(Flow[ConversationState]):
|
|
def route_turn(self, context: dict[str, Any]) -> str:
|
|
return "order"
|
|
|
|
@listen("order")
|
|
def handle_order(self) -> Any:
|
|
# Echoing the route label is a routing artefact, not a reply.
|
|
return TestHandlerReplyPromotion._AgentOutput("order")
|
|
|
|
flow = RoutedChat()
|
|
flow.handle_turn("where is my order?")
|
|
|
|
assert flow.state.last_intent == "order"
|
|
assert [m.role for m in flow.state.messages] == ["user"]
|
|
|
|
def test_non_text_output_is_not_promoted(self) -> None:
|
|
flow = self._chat(self._AgentOutput(raw=None)) # type: ignore[arg-type]
|
|
|
|
flow.handle_turn("hi")
|
|
|
|
assert [m.role for m in flow.state.messages] == ["user"]
|
|
|
|
def test_handler_that_already_replied_is_not_double_promoted(self) -> None:
|
|
@ConversationConfig()
|
|
class ExplicitReplyChat(Flow[ConversationState]):
|
|
@listen("converse")
|
|
def converse_turn(self) -> Any:
|
|
self.append_assistant_message("explicit")
|
|
return TestHandlerReplyPromotion._AgentOutput("returned")
|
|
|
|
flow = ExplicitReplyChat()
|
|
flow.handle_turn("hi")
|
|
|
|
assert [(m.role, m.content) for m in flow.state.messages] == [
|
|
("user", "hi"),
|
|
("assistant", "explicit"),
|
|
]
|