Lorenze/imp/streaming-conversational#6360
Conversation
- Introduced function to determine if a result is a streaming output. - Added method to handle streaming results before accessing them. - Updated method to utilize the new streaming result handling. - Implemented context management for LLM streaming in the conversational mixin. - Added tests to verify streaming behavior and ensure proper handling of user messages during streaming.
![corridor-security[bot]](https://avatars.githubusercontent.com/in/1177339?s=60&v=4){kind=small}
There was a problem hiding this comment.
Summary: This PR changes conversational flow streaming behavior and TUI handling of streaming results, without adding new public endpoints, authentication paths, file/SQL/network inputs, or privilege boundaries. No exploitable security vulnerabilities were identified.
Risk: Low risk. The changes operate on internal flow/runtime state and local CLI result consumption, with no attacker-controlled security-sensitive data handling introduced.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughThe PR adds CLI streaming-output handling and updates conversational flow runtime code to enable LLM streaming during turns, capture and restore pending kickoff context, and wrap sync and async kickoff execution with streaming-run state. ChangesStreaming conversation and kickoff flow
Sequence Diagram(s)sequenceDiagram
participant Flow
participant _ConversationalMixin
participant llm_instance
Flow->>Flow: _capture_pending_kickoff_context()
Flow->>Flow: _should_stream_llm_calls()
Flow->>_ConversationalMixin: _restore_pending_kickoff_context(context)
Flow->>Flow: enter _streaming_run()
Flow->>llm_instance: call(...)
llm_instance-->>Flow: response
Flow->>_ConversationalMixin: _clear_pending_kickoff_context()
Suggested reviewers
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/crewai/src/crewai/experimental/conversational_mixin.py`:
- Around line 66-78: The _streaming_conversation_llm context manager currently
mutates llm.stream on a shared BaseLLM instance, which can race across
overlapping turns and restore the wrong state. Update the conversational flow so
each turn uses a per-call/per-session LLM instance or another concurrency-safe
streaming mechanism instead of modifying shared state in
_streaming_conversation_llm, and keep the streaming enablement scoped to
call()/acall() usage in the conversational mixin.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 4a28d76e-edba-458e-b824-74ba467b2ab7
📒 Files selected for processing (5)
lib/cli/src/crewai_cli/crew_run_tui.pylib/cli/tests/test_crew_run_tui.pylib/crewai/src/crewai/experimental/conversational_mixin.pylib/crewai/src/crewai/flow/runtime/__init__.pylib/crewai/tests/test_flow_conversation.py
| @contextmanager | ||
| def _streaming_conversation_llm(llm: BaseLLM, *, enabled: bool) -> Any: | ||
| """Temporarily enable LLM streaming for Flow streaming turns.""" | ||
| if not enabled: | ||
| yield | ||
| return | ||
|
|
||
| previous_stream = llm.stream | ||
| llm.stream = True | ||
| try: | ||
| yield | ||
| finally: | ||
| llm.stream = previous_stream |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🌐 Web query:
Does the crewAI BaseLLM / LLM.call API accept a per-call streaming parameter, or is streaming controlled only via the instance-level llm.stream attribute?
💡 Result:
In crewAI, streaming is controlled at the instance level through the stream attribute of the LLM object, rather than through a per-call parameter in the call() or acall() API [1][2]. When initializing an LLM instance, you enable streaming by setting stream=True [1][2]: llm = LLM( model="openai/gpt-4o", stream=True) The call() and acall() methods do not accept a streaming flag [3]. Instead, when stream is set to True on the instance, crewAI automatically handles the streaming process by emitting LLMStreamChunkEvent objects via the crewai_event_bus for every token received [4][3][5][6]. To consume these streamed tokens, you must register an event handler with the event bus to listen for these LLMStreamChunkEvent occurrences [4][5][2]. Top results: [1][3][5][2][6]
Citations:
- 1: https://docs.crewai.com/en/concepts/llms
- 2: https://github.com/crewAIInc/crewAI/blob/main/docs/en/concepts/llms.mdx
- 3: https://github.com/crewAIInc/crewAI/blob/e21c5062/lib/crewai/src/crewai/llms/base_llm.py
- 4: https://community.crewai.com/t/openai-streaming/3556
- 5: [FEATURE] Streaming output support #2950
- 6: https://deepwiki.com/crewAIInc/crewAI/4.1-llm-architecture-and-providers
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant files and symbols with line numbers.
git ls-files lib/crewai/src/crewai/experimental/conversational_mixin.py lib/crewai/src/crewai/llms/base_llm.py lib/crewai/src/crewai/llms/llm.py 2>/dev/null || true
printf '\n--- conversational_mixin.py outline ---\n'
ast-grep outline lib/crewai/src/crewai/experimental/conversational_mixin.py --view expanded
printf '\n--- base_llm.py outline ---\n'
ast-grep outline lib/crewai/src/crewai/llms/base_llm.py --view expanded
printf '\n--- search for stream usage and call signatures ---\n'
rg -n "def (call|acall)\(|stream\s*=|stream:" lib/crewai/src/crewai/llms lib/crewai/src/crewai/experimental -g '*.py'Repository: crewAIInc/crewAI
Length of output: 15588
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'FILE: conversational_mixin.py\n'
sed -n '1,120p' lib/crewai/src/crewai/experimental/conversational_mixin.py | cat -n
printf '\nFILE: base_llm.py\n'
sed -n '1,260p' lib/crewai/src/crewai/llms/base_llm.py | cat -n
printf '\nFILE: llm.py\n'
sed -n '1,260p' lib/crewai/src/crewai/llms/llm.py | cat -nRepository: crewAIInc/crewAI
Length of output: 15468
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the relevant files first, then read only the needed sections.
printf '--- conversational_mixin.py outline ---\n'
ast-grep outline lib/crewai/src/crewai/experimental/conversational_mixin.py --view expanded
printf '\n--- conversational_mixin.py lines 1-220 ---\n'
sed -n '1,220p' lib/crewai/src/crewai/experimental/conversational_mixin.py | cat -n
printf '\n--- conversational_mixin.py lines 220-820 ---\n'
sed -n '220,820p' lib/crewai/src/crewai/experimental/conversational_mixin.py | cat -n
printf '\n--- base_llm.py outline ---\n'
ast-grep outline lib/crewai/src/crewai/llms/base_llm.py --view expanded
printf '\n--- base_llm.py lines 1-260 ---\n'
sed -n '1,260p' lib/crewai/src/crewai/llms/base_llm.py | cat -nRepository: crewAIInc/crewAI
Length of output: 245
🏁 Script executed:
#!/bin/bash
set -euo pipefail
set +e
printf '%s\n' '--- conversational_mixin.py around _default_conversation_llm ---'
sed -n '1010,1050p' lib/crewai/src/crewai/experimental/conversational_mixin.py | cat -n
printf '\n%s\n' '--- conversational_mixin.py around converse_turn ---'
sed -n '220,260p' lib/crewai/src/crewai/experimental/conversational_mixin.py | cat -n
printf '\n%s\n' '--- base_llm.py around call/acall ---'
sed -n '280,360p' lib/crewai/src/crewai/llms/base_llm.py | cat -n
printf '\n%s\n' '--- base_llm.py around stream handling helper ---'
sed -n '480,550p' lib/crewai/src/crewai/llms/base_llm.py | cat -nRepository: crewAIInc/crewAI
Length of output: 11071
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find where conversational_config / llm are assigned and used.
rg -n "conversational_config|_default_conversation_llm\(|llm =" lib/crewai/src/crewai/experimental/conversational_mixin.py lib/crewai/src/crewai/experimental -g '*.py'
# Inspect the ConversationConfig definition to see whether llm is mutable/shared by design.
rg -n "class ConversationConfig|llm:|router_llm:|config.llm" lib/crewai/src/crewai/experimental -g '*.py'Repository: crewAIInc/crewAI
Length of output: 5604
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- conversational_mixin.py around _default_conversation_llm ---'
sed -n '1010,1050p' lib/crewai/src/crewai/experimental/conversational_mixin.py | cat -n
printf '\n%s\n' '--- conversational_mixin.py around converse_turn ---'
sed -n '220,260p' lib/crewai/src/crewai/experimental/conversational_mixin.py | cat -n
printf '\n%s\n' '--- base_llm.py around call/acall ---'
sed -n '280,360p' lib/crewai/src/crewai/llms/base_llm.py | cat -n
printf '\n%s\n' '--- base_llm.py around stream handling helper ---'
sed -n '480,550p' lib/crewai/src/crewai/llms/base_llm.py | cat -n
printf '\n%s\n' '--- conversational config and llm assignment/search ---'
rg -n "class ConversationConfig|conversational_config|_default_conversation_llm\(|config\.llm|llm =" lib/crewai/src/crewai/experimental -g '*.py'Repository: crewAIInc/crewAI
Length of output: 13283
Avoid mutating llm.stream on a shared instance
config.llm comes from the class-level conversational_config, so overlapping turns/sessions can race on this save/restore and leave stream in the wrong state. call()/acall() don’t accept a per-call streaming flag, so this needs a per-turn LLM instance or another concurrency-safe streaming hook instead of mutating shared state.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/crewai/src/crewai/experimental/conversational_mixin.py` around lines 66 -
78, The _streaming_conversation_llm context manager currently mutates llm.stream
on a shared BaseLLM instance, which can race across overlapping turns and
restore the wrong state. Update the conversational flow so each turn uses a
per-call/per-session LLM instance or another concurrency-safe streaming
mechanism instead of modifying shared state in _streaming_conversation_llm, and
keep the streaming enablement scoped to call()/acall() usage in the
conversational mixin.
2 participants
Note
Medium Risk
Changes nested kickoff and experimental conversational streaming paths; behavior is covered by new tests but affects core flow execution timing and LLM stream flags.
Overview
Adds streaming support for conversational Flow turns when
stream=True, including LLM chunk emission during built-inconversehandling and correct behavior whenhandle_turnreturns aFlowStreamingOutput.The flow runtime now tracks an active streaming run (
_streaming_run/_should_stream_llm_calls) so inner kickoff can stream without relying on togglingself.streamalone. For streaming kickoff, pending conversational kickoff context (user message, intents, intent LLM) is captured before the background run, restored inside the worker, and cleared infinally—fixing lost user messages whenhandle_turnclears pending state before the deferred thread executes.The CLI crew-run TUI drains streaming turn results via
_consume_conversation_streaming_resultbefore reading.result, avoiding premature access errors.Reviewed by Cursor Bugbot for commit 4fd8327. Bugbot is set up for automated code reviews on this repo. Configure here.
Summary by CodeRabbit
New Features
Bug Fixes
Tests