class LitellmStopWordLLM(LLM):
"""CrewAI LLM subclass that forces LiteLLM usage and enforces client-side stop-word truncation.
CrewAI's ``LLM.__new__`` may choose a native client instead of LiteLLM for some
model strings. The ``__new__`` override forces ``object.__new__`` so that LiteLLM
is always used. The ``call()`` override ensures stop words are honoured even when
the underlying API silently ignores the stop parameter.
"""
def __new__(cls, *args: Any, **kwargs: Any) -> "LitellmStopWordLLM":
return object.__new__(cls)
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.is_litellm = True
def _collect_chunk(
self,
chunk: Any,
text: list[str],
tool_calls: list[Any],
call_id: str,
callbacks: list | None,
) -> Any:
"""Gather a chunk's content + tool-call parts; return its usage (skips empty choices)."""
if chunk.choices:
delta = chunk.choices[0].delta
if delta.content:
text.append(delta.content)
event = LLMStreamChunkEvent(chunk=delta.content, call_id=call_id)
crewai_event_bus.emit(self, event=event)
for cb in callbacks or []:
if hasattr(cb, "on_llm_new_token"):
cb.on_llm_new_token(delta.content)
if getattr(delta, "tool_calls", None):
tool_calls.extend(delta.tool_calls)
usage = getattr(chunk, "usage", None)
return usage if not isinstance(usage, type) else None
def _track_native_usage(self, usage: Any) -> None:
"""Track streamed usage in CrewAI's metrics (the native loop bypasses the base path)."""
if usage:
self._track_token_usage_internal(self._usage_to_dict(usage) or {})
def _finalize_native(self, text: list[str], tool_calls: list[Any]) -> list[dict] | str:
"""Return streamed tool calls as the bare list CrewAI runs, else truncated text."""
if tool_calls:
from datarobot_genai.core.router import merge_streaming_tool_calls # noqa: PLC0415
return merge_streaming_tool_calls(tool_calls)
return self._apply_stop_words("".join(text))
def _apply_stop_words(self, content: str) -> str:
"""Apply configured stop words, then truncate inline ReAct hallucinations."""
truncated = super()._apply_stop_words(content)
return self._truncate_react_hallucination_after_action_input(truncated)
@staticmethod
def _truncate_react_hallucination_after_action_input(content: str) -> str:
"""Truncate hallucinated text appended after ``Action Input:``.
Models sometimes emit fake tool results or a second ReAct step inline,
without an ``Observation:`` label. Keep only the action-input value:
text on the same line as ``Action Input:``, stopping before any inline
``Thought:`` or ``Final Answer:`` label.
"""
marker = "Action Input:"
marker_start = content.find(marker)
if marker_start == -1:
return content
value_start = marker_start + len(marker)
value_text = content[value_start:]
# Earliest index in ``content`` where hallucinated suffix may begin.
truncation_points = [len(content)]
newline_in_value = value_text.find("\n")
if newline_in_value != -1:
truncation_points.append(value_start + newline_in_value)
for react_label in ("Thought:", "Final Answer:"):
label_offset = value_text.find(react_label)
# Ignore labels glued directly to the marker (offset 0).
if label_offset > 0:
truncation_points.append(value_start + label_offset)
cut_end = min(truncation_points)
if cut_end == len(content):
return content
return content[:cut_end].rstrip()
@staticmethod
def _wants_native_tool_calls(kwargs: dict) -> bool:
"""CrewAI's native loop calls us with ``tools`` set and ``available_functions=None``."""
return bool(kwargs.get("tools")) and kwargs.get("available_functions") is None
def call(self, *args: Any, **kwargs: Any) -> Any:
"""Stream and return native tool calls ourselves — CrewAI's handler drops them.
Non-native calls delegate to the base, stop-word-truncating any text result.
"""
if self._wants_native_tool_calls(kwargs):
import litellm # noqa: PLC0415
with _llm_span(self.model):
tools = _sanitize_tool_schema(kwargs["tools"])
params = self._prepare_completion_params(args[0], tools)
params["stream"] = True
call_id = str(uuid.uuid4())
text: list[str] = []
tool_calls: list[Any] = []
usage: Any = None
try:
for chunk in litellm.completion(**params):
usage = (
self._collect_chunk(
chunk, text, tool_calls, call_id, kwargs.get("callbacks")
)
or usage
)
except Exception as exc:
crewai_event_bus.emit(
self, event=LLMCallFailedEvent(call_id=call_id, error=str(exc))
)
raise
self._track_native_usage(usage)
return self._finalize_native(text, tool_calls)
result = super().call(*args, **kwargs)
return self._apply_stop_words(result) if isinstance(result, str) else result
def _format_messages_for_provider(self, messages: list) -> list:
"""Ensure conversation does not end with an assistant message.
Some models routed through the DataRobot LLM Gateway (e.g. claude-sonnet-4-6)
reject assistant-message prefill. When the conversation ends with an assistant
message we append a minimal user message so the API accepts the request while
preserving the full conversation context.
"""
formatted = super()._format_messages_for_provider(messages)
if formatted and formatted[-1].get("role") == "assistant":
formatted = [*formatted, {"role": "user", "content": "Please continue."}]
return formatted
async def acall(self, *args: Any, **kwargs: Any) -> Any:
"""Async variant of :meth:`call` used by ``Crew.akickoff``."""
if self._wants_native_tool_calls(kwargs):
import litellm # noqa: PLC0415
with _llm_span(self.model):
tools = _sanitize_tool_schema(kwargs["tools"])
params = self._prepare_completion_params(args[0], tools)
params["stream"] = True
call_id = str(uuid.uuid4())
text: list[str] = []
tool_calls: list[Any] = []
usage: Any = None
try:
async for chunk in await litellm.acompletion(**params):
usage = (
self._collect_chunk(
chunk, text, tool_calls, call_id, kwargs.get("callbacks")
)
or usage
)
except Exception as exc:
crewai_event_bus.emit(
self, event=LLMCallFailedEvent(call_id=call_id, error=str(exc))
)
raise
self._track_native_usage(usage)
return self._finalize_native(text, tool_calls)
result = await super().acall(*args, **kwargs)
return self._apply_stop_words(result) if isinstance(result, str) else result
def supports_function_calling(self) -> bool:
supported = _model_supports_tool_calling(self.model)
return supported if supported is not None else super().supports_function_calling()