diff --git a/lib/crewai/src/crewai/events/types/llm_events.py b/lib/crewai/src/crewai/events/types/llm_events.py index b138f908c..b1be8c3d5 100644 --- a/lib/crewai/src/crewai/events/types/llm_events.py +++ b/lib/crewai/src/crewai/events/types/llm_events.py @@ -1,7 +1,7 @@ from enum import Enum from typing import Any, Literal -from pydantic import BaseModel +from pydantic import BaseModel, field_validator from crewai.events.base_events import BaseEvent @@ -48,6 +48,43 @@ class LLMCallStartedEvent(LLMEventBase): tools: list[dict[str, Any]] | None = None callbacks: list[Any] | None = None available_functions: dict[str, Any] | None = None + # Sampling/request parameters forwarded for OTel GenAI compliance. + # All optional so legacy emitters keep working unchanged. + temperature: float | None = None + top_p: float | None = None + max_tokens: int | float | None = None + stream: bool | None = None + seed: int | None = None + stop_sequences: list[str] | None = None + frequency_penalty: float | None = None + presence_penalty: float | None = None + n: int | None = None + + @field_validator("stop_sequences", mode="before") + @classmethod + def _coerce_stop_sequences_to_str_list(cls, value: Any) -> list[str] | None: + """Normalize stop_sequences to ``list[str] | None``. + + Some providers store stop sequences in non-Python-list containers — + e.g. a Vertex AI / Gemini code path can hand back a + ``google.protobuf.struct_pb2.ListValue`` or a ``RepeatedScalarContainer``. + Without coercion the OTel SDK falls back to ``str(value)`` when + ``gen_ai.request.stop_sequences`` is set, producing the protobuf + textproto repr (``values { string_value: \"...\" }``) instead of a + proper ``Sequence[str]``. + + A bare string is treated as a single stop sequence. Anything that + can't be iterated cleanly falls back to ``None`` rather than crashing + event construction. + """ + if value is None: + return None + if isinstance(value, str): + return [value] + try: + return [item if isinstance(item, str) else str(item) for item in value] + except TypeError: + return None class LLMCallCompletedEvent(LLMEventBase): @@ -58,6 +95,23 @@ class LLMCallCompletedEvent(LLMEventBase): response: Any call_type: LLMCallType usage: dict[str, Any] | None = None + finish_reason: str | None = None + response_id: str | None = None + + @field_validator("finish_reason", "response_id", mode="before") + @classmethod + def _coerce_non_string_to_none(cls, value: Any) -> str | None: + """Drop non-string values so test mocks and exotic provider types + (MagicMock, protobuf enums, etc.) never crash event construction. + + Provider helpers are best-effort: when extraction returns something + non-string (e.g. a ``MagicMock`` in unit tests), we treat it as + "no value" rather than raising. Downstream telemetry already + handles the missing-attribute case. + """ + if value is None or isinstance(value, str): + return value + return None class LLMCallFailedEvent(LLMEventBase): diff --git a/lib/crewai/src/crewai/llm.py b/lib/crewai/src/crewai/llm.py index 08c1a1bf8..af5dff68e 100644 --- a/lib/crewai/src/crewai/llm.py +++ b/lib/crewai/src/crewai/llm.py @@ -23,7 +23,6 @@ from crewai.events.event_bus import crewai_event_bus from crewai.events.types.llm_events import ( LLMCallCompletedEvent, LLMCallFailedEvent, - LLMCallStartedEvent, LLMCallType, LLMStreamChunkEvent, ) @@ -32,6 +31,7 @@ from crewai.events.types.tool_usage_events import ( ToolUsageFinishedEvent, ToolUsageStartedEvent, ) +from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id from crewai.llms.base_llm import ( BaseLLM, JsonResponseFormat, @@ -732,6 +732,11 @@ class LLM(BaseLLM): last_chunk = None chunk_count = 0 usage_info = None + # Tracked across the loop: LiteLLM with include_usage emits a final + # usage-only chunk with empty choices, so the post-loop last_chunk has + # no finish_reason. Capture both incrementally instead. + stream_finish_reason: str | None = None + stream_response_id: str | None = None accumulated_tool_args: defaultdict[int, AccumulatedToolArgs] = defaultdict( AccumulatedToolArgs @@ -750,6 +755,16 @@ class LLM(BaseLLM): if isinstance(chunk, ModelResponseBase): response_id = chunk.id + elif isinstance(chunk, dict): + response_id = chunk.get("id") + + chunk_finish, chunk_id = self._extract_finish_reason_and_response_id( + chunk + ) + if chunk_finish: + stream_finish_reason = chunk_finish + if chunk_id and not stream_response_id: + stream_response_id = chunk_id try: choices = None @@ -922,6 +937,11 @@ class LLM(BaseLLM): if tool_calls_list: return tool_calls_list + finish_reason, response_id_last = ( + stream_finish_reason, + stream_response_id, + ) + if not tool_calls or not available_functions: if response_model and self.is_litellm: instructor_instance = InternalInstructor( @@ -939,6 +959,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_dict, + finish_reason=finish_reason, + response_id=response_id_last, ) return structured_response @@ -950,6 +972,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_dict, + finish_reason=finish_reason, + response_id=response_id_last, ) return full_response @@ -965,6 +989,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_dict, + finish_reason=finish_reason, + response_id=response_id_last, ) return full_response @@ -978,6 +1004,10 @@ class LLM(BaseLLM): logging.error(f"Error in streaming response: {e!s}") if full_response.strip(): logging.warning(f"Returning partial response despite error: {e!s}") + finish_reason, response_id_last = ( + stream_finish_reason, + stream_response_id, + ) self._handle_emit_call_events( response=full_response, call_type=LLMCallType.LLM_CALL, @@ -985,6 +1015,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=self._usage_to_dict(usage_info), + finish_reason=finish_reason, + response_id=response_id_last, ) return full_response @@ -1169,6 +1201,10 @@ class LLM(BaseLLM): else None ) + finish_reason, response_id = self._extract_finish_reason_and_response_id( + response + ) + if response_model is not None: # When using instructor/response_model, litellm returns a Pydantic model instance if isinstance(response, BaseModel): @@ -1180,6 +1216,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=response_usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_response @@ -1216,6 +1254,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=response_usage, + finish_reason=finish_reason, + response_id=response_id, ) return text_response @@ -1233,6 +1273,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=response_usage, + finish_reason=finish_reason, + response_id=response_id, ) return text_response @@ -1310,6 +1352,10 @@ class LLM(BaseLLM): else None ) + finish_reason, response_id = self._extract_finish_reason_and_response_id( + response + ) + if response_model is not None: if isinstance(response, BaseModel): structured_response = response.model_dump_json() @@ -1320,6 +1366,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=response_usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_response @@ -1358,6 +1406,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=response_usage, + finish_reason=finish_reason, + response_id=response_id, ) return text_response @@ -1375,6 +1425,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=response_usage, + finish_reason=finish_reason, + response_id=response_id, ) return text_response @@ -1412,12 +1464,29 @@ class LLM(BaseLLM): params["stream"] = True params["stream_options"] = {"include_usage": True} response_id = None + # See sync sibling: incrementally track finish_reason/response_id so the + # usage-only final chunk doesn't wipe them. + stream_finish_reason: str | None = None + stream_response_id: str | None = None try: async for chunk in await litellm.acompletion(**params): chunk_count += 1 chunk_content = None - response_id = chunk.id if isinstance(chunk, ModelResponseBase) else None + if isinstance(chunk, ModelResponseBase): + response_id = chunk.id + elif isinstance(chunk, dict): + response_id = chunk.get("id") + else: + response_id = None + + chunk_finish, chunk_id = self._extract_finish_reason_and_response_id( + chunk + ) + if chunk_finish: + stream_finish_reason = chunk_finish + if chunk_id and not stream_response_id: + stream_response_id = chunk_id try: choices = None @@ -1525,6 +1594,10 @@ class LLM(BaseLLM): return tool_calls_list usage_dict = self._usage_to_dict(usage_info) + finish_reason, response_id_last = ( + stream_finish_reason, + stream_response_id, + ) self._handle_emit_call_events( response=full_response, call_type=LLMCallType.LLM_CALL, @@ -1532,6 +1605,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params.get("messages"), usage=usage_dict, + finish_reason=finish_reason, + response_id=response_id_last, ) return full_response @@ -1545,6 +1620,10 @@ class LLM(BaseLLM): if chunk_count == 0: raise if full_response: + finish_reason, response_id_last = ( + stream_finish_reason, + stream_response_id, + ) self._handle_emit_call_events( response=full_response, call_type=LLMCallType.LLM_CALL, @@ -1552,6 +1631,8 @@ class LLM(BaseLLM): from_agent=from_agent, messages=params.get("messages"), usage=self._usage_to_dict(usage_info), + finish_reason=finish_reason, + response_id=response_id_last, ) return full_response raise @@ -1678,19 +1759,14 @@ class LLM(BaseLLM): ValueError: If response format is not supported LLMContextLengthExceededError: If input exceeds model's context limit """ - with llm_call_context() as call_id: - crewai_event_bus.emit( - self, - event=LLMCallStartedEvent( - messages=messages, - tools=tools, - callbacks=callbacks, - available_functions=available_functions, - from_task=from_task, - from_agent=from_agent, - model=self.model, - call_id=call_id, - ), + with llm_call_context(): + self._emit_call_started_event( + messages=messages, + tools=tools, + callbacks=callbacks, + available_functions=available_functions, + from_task=from_task, + from_agent=from_agent, ) self._validate_call_params() @@ -1822,19 +1898,14 @@ class LLM(BaseLLM): ValueError: If response format is not supported LLMContextLengthExceededError: If input exceeds model's context limit """ - with llm_call_context() as call_id: - crewai_event_bus.emit( - self, - event=LLMCallStartedEvent( - messages=messages, - tools=tools, - callbacks=callbacks, - available_functions=available_functions, - from_task=from_task, - from_agent=from_agent, - model=self.model, - call_id=call_id, - ), + with llm_call_context(): + self._emit_call_started_event( + messages=messages, + tools=tools, + callbacks=callbacks, + available_functions=available_functions, + from_task=from_task, + from_agent=from_agent, ) self._validate_call_params() @@ -1990,6 +2061,8 @@ class LLM(BaseLLM): from_agent: BaseAgent | None = None, messages: str | list[LLMMessage] | None = None, usage: dict[str, Any] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> None: """Handle the events for the LLM call. @@ -2000,6 +2073,10 @@ class LLM(BaseLLM): from_agent: Optional agent object messages: Optional messages object usage: Optional token usage data + finish_reason: Raw provider finish reason (e.g. "stop", "length", + "tool_calls"). Optional; downstream telemetry coerces to the + OTel GenAI enum. + response_id: Raw provider response identifier. Optional. """ crewai_event_bus.emit( self, @@ -2012,9 +2089,24 @@ class LLM(BaseLLM): model=self.model, call_id=get_current_call_id(), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ), ) + def _effective_max_tokens(self) -> int | float | None: + """LiteLLM sends ``max_tokens or max_completion_tokens`` as the cap.""" + return self.max_tokens or self.max_completion_tokens + + @staticmethod + def _extract_finish_reason_and_response_id( + response_or_chunk: Any, + ) -> tuple[str | None, str | None]: + """LiteLLM responses/chunks share the choices-shape with OpenAI/Azure; + delegate to the shared extractor. + """ + return extract_choices_finish_reason_and_id(response_or_chunk) + def _process_message_files(self, messages: list[LLMMessage]) -> list[LLMMessage]: """Process files attached to messages and format for provider. diff --git a/lib/crewai/src/crewai/llms/_finish_reason_utils.py b/lib/crewai/src/crewai/llms/_finish_reason_utils.py new file mode 100644 index 000000000..e79befcc7 --- /dev/null +++ b/lib/crewai/src/crewai/llms/_finish_reason_utils.py @@ -0,0 +1,55 @@ +"""Shared extractors for ``finish_reason`` + ``response_id`` across LLM providers. + +OpenAI Chat Completions, Azure AI Inference, and LiteLLM all expose the same +choices-based response shape (``response.id`` + ``response.choices[0].finish_reason``), +both as object attributes and (for LiteLLM stream chunks) as dict keys. This +module centralises that introspection so every provider doesn't reinvent the +defensive walk. Providers with genuinely different shapes — Anthropic +(``stop_reason``), Bedrock (``stopReason``), Gemini (protobuf enum), OpenAI +Responses (``status``) — keep their own helpers. +""" + +from __future__ import annotations + +from typing import Any + + +def _as_str(value: Any) -> str | None: + return value if isinstance(value, str) else None + + +def extract_choices_finish_reason_and_id( + response_or_chunk: Any, +) -> tuple[str | None, str | None]: + """Extract ``(finish_reason, response_id)`` from a choices-shaped response. + + Handles both object-style (``response.id``, ``response.choices[0].finish_reason``) + and dict-style (``response["id"]``, ``response["choices"][0]["finish_reason"]``) + inputs. Returns ``(None, None)`` on any failure; never raises. Non-string + raw values are coerced to ``None`` so test mocks and exotic provider types + (MagicMock, protobuf enums, etc.) don't propagate downstream. + """ + raw_id = getattr(response_or_chunk, "id", None) + if raw_id is None and isinstance(response_or_chunk, dict): + raw_id = response_or_chunk.get("id") + response_id = _as_str(raw_id) + + if isinstance(response_or_chunk, dict): + choices = response_or_chunk.get("choices") + else: + choices = getattr(response_or_chunk, "choices", None) + + finish_reason: str | None = None + if choices: + try: + first = choices[0] + except (IndexError, TypeError, KeyError): + first = None + if first is not None: + if isinstance(first, dict): + raw_finish = first.get("finish_reason") + else: + raw_finish = getattr(first, "finish_reason", None) + finish_reason = _as_str(raw_finish) + + return finish_reason, response_id diff --git a/lib/crewai/src/crewai/llms/base_llm.py b/lib/crewai/src/crewai/llms/base_llm.py index 83429cdf1..03f277855 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -150,6 +150,13 @@ class BaseLLM(BaseModel, ABC): llm_type: str = "base" model: str temperature: float | None = None + top_p: float | None = None + max_tokens: int | float | None = None + stream: bool | None = None + seed: int | None = None + frequency_penalty: float | None = None + presence_penalty: float | None = None + n: int | None = None api_key: str | None = None base_url: str | None = None provider: str = Field(default="openai") @@ -464,6 +471,16 @@ class BaseLLM(BaseModel, ABC): """ return None + def _effective_max_tokens(self) -> int | float | None: + """Token cap actually sent to the provider, for start-event telemetry. + + Defaults to ``self.max_tokens``. Providers that cap generation through a + differently named field (e.g. ``max_completion_tokens`` on OpenAI/Azure, + ``max_output_tokens`` on Gemini) override this so ``LLMCallStartedEvent`` + reports the real limit instead of ``None``. + """ + return self.max_tokens + def _emit_call_started_event( self, messages: str | list[LLMMessage], @@ -472,10 +489,38 @@ class BaseLLM(BaseModel, ABC): available_functions: dict[str, Any] | None = None, from_task: Task | None = None, from_agent: BaseAgent | None = None, + temperature: float | None = None, + top_p: float | None = None, + max_tokens: int | float | None = None, + stream: bool | None = None, + seed: int | None = None, + stop_sequences: list[str] | None = None, + frequency_penalty: float | None = None, + presence_penalty: float | None = None, + n: int | None = None, ) -> None: """Emit LLM call started event.""" from crewai.utilities.serialization import to_serializable + if temperature is None: + temperature = self.temperature + if top_p is None: + top_p = self.top_p + if max_tokens is None: + max_tokens = self._effective_max_tokens() + if stream is None: + stream = self.stream + if seed is None: + seed = self.seed + if stop_sequences is None: + stop_sequences = self.stop_sequences or None + if frequency_penalty is None: + frequency_penalty = self.frequency_penalty + if presence_penalty is None: + presence_penalty = self.presence_penalty + if n is None: + n = self.n + crewai_event_bus.emit( self, event=LLMCallStartedEvent( @@ -487,6 +532,15 @@ class BaseLLM(BaseModel, ABC): from_agent=from_agent, model=self.model, call_id=get_current_call_id(), + temperature=temperature, + top_p=top_p, + max_tokens=max_tokens, + stream=stream, + seed=seed, + stop_sequences=stop_sequences, + frequency_penalty=frequency_penalty, + presence_penalty=presence_penalty, + n=n, ), ) @@ -498,6 +552,8 @@ class BaseLLM(BaseModel, ABC): from_agent: BaseAgent | None = None, messages: str | list[LLMMessage] | None = None, usage: dict[str, Any] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> None: """Emit LLM call completed event.""" from crewai.utilities.serialization import to_serializable @@ -513,6 +569,8 @@ class BaseLLM(BaseModel, ABC): model=self.model, call_id=get_current_call_id(), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ), ) diff --git a/lib/crewai/src/crewai/llms/providers/anthropic/completion.py b/lib/crewai/src/crewai/llms/providers/anthropic/completion.py index 28122d4db..599ec5a3b 100644 --- a/lib/crewai/src/crewai/llms/providers/anthropic/completion.py +++ b/lib/crewai/src/crewai/llms/providers/anthropic/completion.py @@ -923,6 +923,8 @@ class AnthropicCompletion(BaseLLM): usage = self._extract_anthropic_token_usage(response) self._track_token_usage_internal(usage) + finish_reason, response_id = self._extract_finish_reason_and_id(response) + if _is_pydantic_model_class(response_model) and response.content: if use_native_structured_output: for block in response.content: @@ -935,6 +937,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_data else: @@ -951,6 +955,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_data @@ -973,6 +979,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return list(tool_uses) @@ -1005,6 +1013,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) if usage.get("total_tokens", 0) > 0: @@ -1147,6 +1157,10 @@ class AnthropicCompletion(BaseLLM): usage = self._extract_anthropic_token_usage(final_message) self._track_token_usage_internal(usage) + finish_reason, final_response_id = self._extract_finish_reason_and_id( + final_message + ) + if _is_pydantic_model_class(response_model): if use_native_structured_output: structured_data = response_model.model_validate_json(full_response) @@ -1157,6 +1171,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=final_response_id, ) return structured_data for block in final_message.content: @@ -1172,6 +1188,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=final_response_id, ) return structured_data @@ -1201,6 +1219,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=final_response_id, ) return self._invoke_after_llm_call_hooks( @@ -1361,6 +1381,10 @@ class AnthropicCompletion(BaseLLM): final_content = self._apply_stop_words(final_content) + finish_reason, final_response_id = self._extract_finish_reason_and_id( + final_response + ) + self._emit_call_completed_event( response=final_content, call_type=LLMCallType.LLM_CALL, @@ -1368,6 +1392,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=follow_up_params["messages"], usage=follow_up_usage, + finish_reason=finish_reason, + response_id=final_response_id, ) total_usage = { @@ -1447,6 +1473,8 @@ class AnthropicCompletion(BaseLLM): usage = self._extract_anthropic_token_usage(response) self._track_token_usage_internal(usage) + finish_reason, response_id = self._extract_finish_reason_and_id(response) + if _is_pydantic_model_class(response_model) and response.content: if use_native_structured_output: for block in response.content: @@ -1459,6 +1487,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_data else: @@ -1475,6 +1505,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_data @@ -1495,6 +1527,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return list(tool_uses) @@ -1519,6 +1553,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) if usage.get("total_tokens", 0) > 0: @@ -1647,6 +1683,10 @@ class AnthropicCompletion(BaseLLM): usage = self._extract_anthropic_token_usage(final_message) self._track_token_usage_internal(usage) + finish_reason, final_response_id = self._extract_finish_reason_and_id( + final_message + ) + if _is_pydantic_model_class(response_model): if use_native_structured_output: structured_data = response_model.model_validate_json(full_response) @@ -1657,6 +1697,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=final_response_id, ) return structured_data for block in final_message.content: @@ -1672,6 +1714,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=final_response_id, ) return structured_data @@ -1701,6 +1745,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=final_response_id, ) return full_response @@ -1753,6 +1799,10 @@ class AnthropicCompletion(BaseLLM): final_content = self._apply_stop_words(final_content) + finish_reason, final_response_id = self._extract_finish_reason_and_id( + final_response + ) + self._emit_call_completed_event( response=final_content, call_type=LLMCallType.LLM_CALL, @@ -1760,6 +1810,8 @@ class AnthropicCompletion(BaseLLM): from_agent=from_agent, messages=follow_up_params["messages"], usage=follow_up_usage, + finish_reason=finish_reason, + response_id=final_response_id, ) total_usage = { @@ -1813,6 +1865,20 @@ class AnthropicCompletion(BaseLLM): return int(200000 * CONTEXT_WINDOW_USAGE_RATIO) + @staticmethod + def _extract_finish_reason_and_id( + message: Any, + ) -> tuple[str | None, str | None]: + """Extract raw finish_reason and response_id from an Anthropic + ``Message`` / ``BetaMessage``. Anthropic exposes ``stop_reason`` (e.g. + ``"end_turn"``, ``"max_tokens"``, ``"tool_use"``); we forward it raw + and let downstream telemetry map to the OTel GenAI enum. + """ + return ( + getattr(message, "stop_reason", None), + getattr(message, "id", None), + ) + @staticmethod def _extract_anthropic_token_usage( response: Message | BetaMessage, diff --git a/lib/crewai/src/crewai/llms/providers/azure/completion.py b/lib/crewai/src/crewai/llms/providers/azure/completion.py index d357939bb..618ed5811 100644 --- a/lib/crewai/src/crewai/llms/providers/azure/completion.py +++ b/lib/crewai/src/crewai/llms/providers/azure/completion.py @@ -9,6 +9,7 @@ from urllib.parse import urlparse from pydantic import BaseModel, PrivateAttr, model_validator from typing_extensions import Self +from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id from crewai.llms.hooks.base import BaseInterceptor from crewai.utilities.agent_utils import is_context_length_exceeded from crewai.utilities.exceptions.context_window_exceeding_exception import ( @@ -783,6 +784,8 @@ class AzureCompletion(BaseLLM): from_task: Any | None = None, from_agent: Any | None = None, usage: dict[str, Any] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> BaseModel: """Validate content against response model and emit completion event. @@ -792,6 +795,8 @@ class AzureCompletion(BaseLLM): params: Completion parameters containing messages from_task: Task that initiated the call from_agent: Agent that initiated the call + finish_reason: Raw provider finish reason. + response_id: Raw provider response id. Returns: Validated Pydantic model instance @@ -809,6 +814,8 @@ class AzureCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_data @@ -848,6 +855,8 @@ class AzureCompletion(BaseLLM): usage = self._extract_azure_token_usage(response) self._track_token_usage_internal(usage) + finish_reason, response_id = self._extract_finish_reason_and_id(response) + # Without available_functions, return tool_calls so the caller (executor) handles execution if message.tool_calls and not available_functions: self._emit_call_completed_event( @@ -857,6 +866,8 @@ class AzureCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return list(message.tool_calls) @@ -892,6 +903,8 @@ class AzureCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) content = self._apply_stop_words(content) @@ -903,6 +916,8 @@ class AzureCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return self._invoke_after_llm_call_hooks( @@ -1011,6 +1026,8 @@ class AzureCompletion(BaseLLM): from_task: Any | None = None, from_agent: Any | None = None, response_model: type[BaseModel] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> str | Any: """Finalize streaming response with usage tracking, tool execution, and events. @@ -1039,6 +1056,8 @@ class AzureCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) # Without available_functions, return tool calls in OpenAI-compatible format for the executor @@ -1061,6 +1080,8 @@ class AzureCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) return formatted_tool_calls @@ -1094,6 +1115,8 @@ class AzureCompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) return self._invoke_after_llm_call_hooks( @@ -1113,8 +1136,16 @@ class AzureCompletion(BaseLLM): tool_calls: dict[int, dict[str, Any]] = {} usage_data: dict[str, Any] | None = None + stream_finish_reason: str | None = None + stream_response_id: str | None = None for update in self._get_sync_client().complete(**params): if isinstance(update, StreamingChatCompletionsUpdate): + chunk_finish, chunk_id = self._extract_finish_reason_and_id(update) + if chunk_finish: + stream_finish_reason = chunk_finish + if chunk_id: + stream_response_id = chunk_id + if update.usage: usage = update.usage usage_data = { @@ -1141,6 +1172,8 @@ class AzureCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, response_model=response_model, + finish_reason=stream_finish_reason, + response_id=stream_response_id, ) async def _ahandle_completion( @@ -1180,10 +1213,18 @@ class AzureCompletion(BaseLLM): tool_calls: dict[int, dict[str, Any]] = {} usage_data: dict[str, Any] | None = None + stream_finish_reason: str | None = None + stream_response_id: str | None = None stream = await self._get_async_client().complete(**params) async for update in stream: if isinstance(update, StreamingChatCompletionsUpdate): + chunk_finish, chunk_id = self._extract_finish_reason_and_id(update) + if chunk_finish: + stream_finish_reason = chunk_finish + if chunk_id: + stream_response_id = chunk_id + if hasattr(update, "usage") and update.usage: usage = update.usage usage_data = { @@ -1210,6 +1251,8 @@ class AzureCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, response_model=response_model, + finish_reason=stream_finish_reason, + response_id=stream_response_id, ) def supports_function_calling(self) -> bool: @@ -1271,6 +1314,19 @@ class AzureCompletion(BaseLLM): return int(8192 * CONTEXT_WINDOW_USAGE_RATIO) + def _effective_max_tokens(self) -> int | float | None: + """Azure reasoning/newer chat models cap via ``max_completion_tokens``.""" + return self.max_tokens or self.max_completion_tokens + + @staticmethod + def _extract_finish_reason_and_id( + response_or_update: Any, + ) -> tuple[str | None, str | None]: + """Azure ``ChatCompletions`` / ``StreamingChatCompletionsUpdate`` + share the choices-shape; delegate to the shared extractor. + """ + return extract_choices_finish_reason_and_id(response_or_update) + @staticmethod def _extract_azure_token_usage(response: ChatCompletions) -> dict[str, Any]: """Extract token usage and response metadata from Azure response.""" diff --git a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py index e9790c577..0f34b6723 100644 --- a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py +++ b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py @@ -677,7 +677,7 @@ class BedrockCompletion(BaseLLM): if usage: self._track_token_usage_internal(usage) - stop_reason = response.get("stopReason") + stop_reason, response_id = self._extract_finish_reason_and_id(response) if stop_reason: logging.debug(f"Response stop reason: {stop_reason}") if stop_reason == "max_tokens": @@ -716,6 +716,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage, + finish_reason=stop_reason, + response_id=response_id, ) return result except Exception as e: @@ -738,6 +740,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage, + finish_reason=stop_reason, + response_id=response_id, ) return non_structured_output_tool_uses @@ -812,6 +816,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage, + finish_reason=stop_reason, + response_id=response_id, ) return self._invoke_after_llm_call_hooks( @@ -951,7 +957,9 @@ class BedrockCompletion(BaseLLM): ) stream = response.get("stream") - response_id = None + _, stream_response_id = self._extract_finish_reason_and_id(response) + response_id = stream_response_id + stream_finish_reason: str | None = None if stream: for event in stream: if "messageStart" in event: @@ -1042,6 +1050,9 @@ class BedrockCompletion(BaseLLM): result = response_model.model_validate( function_args ) + # contentBlockStop fires before messageStop sets + # stream_finish_reason; structured output always + # completes via the tool-call path. self._emit_call_completed_event( response=result.model_dump_json(), call_type=LLMCallType.LLM_CALL, @@ -1049,6 +1060,9 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage_data, + finish_reason=stream_finish_reason + or "tool_use", + response_id=response_id, ) return result # type: ignore[return-value] except Exception as e: @@ -1102,6 +1116,7 @@ class BedrockCompletion(BaseLLM): tool_use_id = None elif "messageStop" in event: stop_reason = event["messageStop"].get("stopReason") + stream_finish_reason = stop_reason logging.debug(f"Streaming message stopped: {stop_reason}") if stop_reason == "max_tokens": logging.warning( @@ -1147,6 +1162,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage_data, + finish_reason=stream_finish_reason, + response_id=response_id, ) return full_response @@ -1262,7 +1279,7 @@ class BedrockCompletion(BaseLLM): if usage: self._track_token_usage_internal(usage) - stop_reason = response.get("stopReason") + stop_reason, response_id = self._extract_finish_reason_and_id(response) if stop_reason: logging.debug(f"Response stop reason: {stop_reason}") if stop_reason == "max_tokens": @@ -1300,6 +1317,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage, + finish_reason=stop_reason, + response_id=response_id, ) return result except Exception as e: @@ -1322,6 +1341,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage, + finish_reason=stop_reason, + response_id=response_id, ) return non_structured_output_tool_uses @@ -1397,6 +1418,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage, + finish_reason=stop_reason, + response_id=response_id, ) return text_content @@ -1531,7 +1554,9 @@ class BedrockCompletion(BaseLLM): ) stream = response.get("stream") - response_id = None + _, stream_response_id = self._extract_finish_reason_and_id(response) + response_id = stream_response_id + stream_finish_reason: str | None = None if stream: async for event in stream: if "messageStart" in event: @@ -1623,6 +1648,9 @@ class BedrockCompletion(BaseLLM): result = response_model.model_validate( function_args ) + # contentBlockStop fires before messageStop sets + # stream_finish_reason; structured output always + # completes via the tool-call path. self._emit_call_completed_event( response=result.model_dump_json(), call_type=LLMCallType.LLM_CALL, @@ -1630,6 +1658,9 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage_data, + finish_reason=stream_finish_reason + or "tool_use", + response_id=response_id, ) return result # type: ignore[return-value] except Exception as e: @@ -1687,6 +1718,7 @@ class BedrockCompletion(BaseLLM): elif "messageStop" in event: stop_reason = event["messageStop"].get("stopReason") + stream_finish_reason = stop_reason logging.debug(f"Streaming message stopped: {stop_reason}") if stop_reason == "max_tokens": logging.warning( @@ -1733,6 +1765,8 @@ class BedrockCompletion(BaseLLM): from_agent=from_agent, messages=messages, usage=usage_data, + finish_reason=stream_finish_reason, + response_id=response_id, ) return self._invoke_after_llm_call_hooks( @@ -1988,6 +2022,25 @@ class BedrockCompletion(BaseLLM): return config + @staticmethod + def _extract_finish_reason_and_id( + response: Any, + ) -> tuple[str | None, str | None]: + """Extract raw finish_reason (``stopReason``) from a Bedrock Converse + response dict. Defensive — returns (None, None) on any failure. + + Bedrock Converse has no model-level response id; ResponseMetadata.RequestId + is an AWS infra trace id (semantically different from OpenAI's chatcmpl-XXX), + so we omit response_id rather than mislead downstream telemetry consumers. + """ + finish_reason: str | None = None + try: + if isinstance(response, dict): + finish_reason = response.get("stopReason") + except (AttributeError, KeyError, TypeError, IndexError): + finish_reason = None + return finish_reason, None + def _handle_client_error(self, e: ClientError) -> str: """Handle AWS ClientError with specific error codes and return error message.""" error_code = e.response.get("Error", {}).get("Code", "Unknown") diff --git a/lib/crewai/src/crewai/llms/providers/gemini/completion.py b/lib/crewai/src/crewai/llms/providers/gemini/completion.py index 8914b6b26..b811614a1 100644 --- a/lib/crewai/src/crewai/llms/providers/gemini/completion.py +++ b/lib/crewai/src/crewai/llms/providers/gemini/completion.py @@ -682,6 +682,8 @@ class GeminiCompletion(BaseLLM): from_task: Any | None = None, from_agent: Any | None = None, usage: dict[str, Any] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> BaseModel: """Validate content against response model and emit completion event. @@ -691,6 +693,8 @@ class GeminiCompletion(BaseLLM): messages_for_event: Messages to include in event from_task: Task that initiated the call from_agent: Agent that initiated the call + finish_reason: Raw provider finish reason. + response_id: Raw provider response id. Returns: Validated Pydantic model instance @@ -708,6 +712,8 @@ class GeminiCompletion(BaseLLM): from_agent=from_agent, messages=messages_for_event, usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_data @@ -724,6 +730,8 @@ class GeminiCompletion(BaseLLM): from_task: Any | None = None, from_agent: Any | None = None, usage: dict[str, Any] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> str | BaseModel: """Finalize completion response with validation and event emission. @@ -747,6 +755,8 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) self._emit_call_completed_event( @@ -756,6 +766,8 @@ class GeminiCompletion(BaseLLM): from_agent=from_agent, messages=messages_for_event, usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return self._invoke_after_llm_call_hooks( @@ -770,6 +782,8 @@ class GeminiCompletion(BaseLLM): from_task: Any | None = None, from_agent: Any | None = None, usage: dict[str, Any] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> BaseModel: """Validate and emit event for structured_output tool call. @@ -795,6 +809,8 @@ class GeminiCompletion(BaseLLM): from_agent=from_agent, messages=self._convert_contents_to_dict(contents), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return validated_data except Exception as e: @@ -828,6 +844,8 @@ class GeminiCompletion(BaseLLM): Returns: Final response content or function call result """ + finish_reason, response_id = self._extract_finish_reason_and_id(response) + if response.candidates and (self.tools or available_functions): candidate = response.candidates[0] if candidate.content and candidate.content.parts: @@ -854,6 +872,8 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) non_structured_output_parts = [ @@ -875,6 +895,8 @@ class GeminiCompletion(BaseLLM): from_agent=from_agent, messages=self._convert_contents_to_dict(contents), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return non_structured_output_parts @@ -915,6 +937,8 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) def _process_stream_chunk( @@ -925,7 +949,13 @@ class GeminiCompletion(BaseLLM): usage_data: dict[str, int] | None, from_task: Any | None = None, from_agent: Any | None = None, - ) -> tuple[str, dict[int, dict[str, Any]], dict[str, int] | None]: + ) -> tuple[ + str, + dict[int, dict[str, Any]], + dict[str, int] | None, + str | None, + str | None, + ]: """Process a single streaming chunk. Args: @@ -937,9 +967,13 @@ class GeminiCompletion(BaseLLM): from_agent: Agent that initiated the call Returns: - Tuple of (updated full_response, updated function_calls, updated usage_data) + Tuple of (updated full_response, updated function_calls, updated + usage_data, chunk finish_reason, chunk response_id). """ response_id = chunk.response_id if hasattr(chunk, "response_id") else None + chunk_finish_reason, chunk_response_id = self._extract_finish_reason_and_id( + chunk + ) if chunk.usage_metadata: usage_data = self._extract_token_usage(chunk) @@ -996,7 +1030,13 @@ class GeminiCompletion(BaseLLM): response_id=response_id, ) - return full_response, function_calls, usage_data + return ( + full_response, + function_calls, + usage_data, + chunk_finish_reason, + chunk_response_id, + ) def _finalize_streaming_response( self, @@ -1008,6 +1048,8 @@ class GeminiCompletion(BaseLLM): from_task: Any | None = None, from_agent: Any | None = None, response_model: type[BaseModel] | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> str | BaseModel | list[dict[str, Any]]: """Finalize streaming response with usage tracking, function execution, and events. @@ -1038,6 +1080,8 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) non_structured_output_calls = { @@ -1058,6 +1102,8 @@ class GeminiCompletion(BaseLLM): from_agent=from_agent, messages=self._convert_contents_to_dict(contents), usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) return raw_parts @@ -1095,6 +1141,8 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) def _handle_completion( @@ -1148,6 +1196,8 @@ class GeminiCompletion(BaseLLM): full_response = "" function_calls: dict[int, dict[str, Any]] = {} usage_data: dict[str, int] | None = None + stream_finish_reason: str | None = None + stream_response_id: str | None = None # The API accepts list[Content] but mypy is overly strict about variance contents_for_api: Any = contents @@ -1156,7 +1206,13 @@ class GeminiCompletion(BaseLLM): contents=contents_for_api, config=config, ): - full_response, function_calls, usage_data = self._process_stream_chunk( + ( + full_response, + function_calls, + usage_data, + chunk_finish_reason, + chunk_response_id, + ) = self._process_stream_chunk( chunk=chunk, full_response=full_response, function_calls=function_calls, @@ -1164,6 +1220,10 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, ) + if chunk_finish_reason: + stream_finish_reason = chunk_finish_reason + if chunk_response_id: + stream_response_id = chunk_response_id return self._finalize_streaming_response( full_response=full_response, @@ -1174,6 +1234,8 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, response_model=response_model, + finish_reason=stream_finish_reason, + response_id=stream_response_id, ) async def _ahandle_completion( @@ -1227,6 +1289,8 @@ class GeminiCompletion(BaseLLM): full_response = "" function_calls: dict[int, dict[str, Any]] = {} usage_data: dict[str, int] | None = None + stream_finish_reason: str | None = None + stream_response_id: str | None = None # The API accepts list[Content] but mypy is overly strict about variance contents_for_api: Any = contents @@ -1236,7 +1300,13 @@ class GeminiCompletion(BaseLLM): config=config, ) async for chunk in stream: - full_response, function_calls, usage_data = self._process_stream_chunk( + ( + full_response, + function_calls, + usage_data, + chunk_finish_reason, + chunk_response_id, + ) = self._process_stream_chunk( chunk=chunk, full_response=full_response, function_calls=function_calls, @@ -1244,6 +1314,10 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, ) + if chunk_finish_reason: + stream_finish_reason = chunk_finish_reason + if chunk_response_id: + stream_response_id = chunk_response_id return self._finalize_streaming_response( full_response=full_response, @@ -1254,6 +1328,8 @@ class GeminiCompletion(BaseLLM): from_task=from_task, from_agent=from_agent, response_model=response_model, + finish_reason=stream_finish_reason, + response_id=stream_response_id, ) def supports_function_calling(self) -> bool: @@ -1300,6 +1376,34 @@ class GeminiCompletion(BaseLLM): return int(1048576 * CONTEXT_WINDOW_USAGE_RATIO) # 1M tokens default + def _effective_max_tokens(self) -> int | float | None: + """Gemini caps generation via ``max_output_tokens``.""" + return self.max_output_tokens or self.max_tokens + + @staticmethod + def _extract_finish_reason_and_id( + response: Any, + ) -> tuple[str | None, str | None]: + """Extract raw finish_reason and response_id from a Gemini + ``GenerateContentResponse``. ``finish_reason`` is the protobuf enum's + ``.name`` attribute (e.g. ``"STOP"``, ``"MAX_TOKENS"``); we forward + it raw and let downstream telemetry map to the OTel GenAI enum. + """ + raw_response_id = getattr(response, "response_id", None) + response_id = raw_response_id if isinstance(raw_response_id, str) else None + + finish_reason: str | None = None + candidates = getattr(response, "candidates", None) + if candidates: + try: + candidate_finish = getattr(candidates[0], "finish_reason", None) + except (IndexError, TypeError, KeyError): + candidate_finish = None + if candidate_finish is not None: + name = getattr(candidate_finish, "name", None) + finish_reason = name if isinstance(name, str) else None + return finish_reason, response_id + @staticmethod def _extract_token_usage(response: GenerateContentResponse) -> dict[str, Any]: """Extract token usage and response metadata from Gemini response.""" diff --git a/lib/crewai/src/crewai/llms/providers/openai/completion.py b/lib/crewai/src/crewai/llms/providers/openai/completion.py index 0adcd82d6..4a610423c 100644 --- a/lib/crewai/src/crewai/llms/providers/openai/completion.py +++ b/lib/crewai/src/crewai/llms/providers/openai/completion.py @@ -29,6 +29,7 @@ from openai.types.responses import ( from pydantic import BaseModel, PrivateAttr, model_validator from crewai.events.types.llm_events import LLMCallType +from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id from crewai.llms.base_llm import BaseLLM, JsonResponseFormat, llm_call_context from crewai.llms.hooks.base import BaseInterceptor from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport @@ -825,6 +826,10 @@ class OpenAICompletion(BaseLLM): usage = self._extract_responses_token_usage(response) self._track_token_usage_internal(usage) + finish_reason, response_id = self._extract_responses_finish_reason_and_id( + response + ) + if self.parse_tool_outputs: parsed_result = self._extract_builtin_tool_outputs(response) parsed_result.text = self._apply_stop_words(parsed_result.text) @@ -836,6 +841,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return parsed_result @@ -849,6 +856,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return function_calls @@ -887,6 +896,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_result except ValueError as e: @@ -901,6 +912,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) content = self._invoke_after_llm_call_hooks( @@ -960,6 +973,10 @@ class OpenAICompletion(BaseLLM): usage = self._extract_responses_token_usage(response) self._track_token_usage_internal(usage) + finish_reason, response_id = self._extract_responses_finish_reason_and_id( + response + ) + if self.parse_tool_outputs: parsed_result = self._extract_builtin_tool_outputs(response) parsed_result.text = self._apply_stop_words(parsed_result.text) @@ -971,6 +988,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return parsed_result @@ -984,6 +1003,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return function_calls @@ -1022,6 +1043,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_result except ValueError as e: @@ -1036,6 +1059,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) except NotFoundError as e: @@ -1123,6 +1148,12 @@ class OpenAICompletion(BaseLLM): usage = self._extract_responses_token_usage(event.response) self._track_token_usage_internal(usage) + finish_reason, response_id = ( + self._extract_responses_finish_reason_and_id(final_response) + if final_response is not None + else (None, response_id_stream) + ) + if self.parse_tool_outputs and final_response: parsed_result = self._extract_builtin_tool_outputs(final_response) parsed_result.text = self._apply_stop_words(parsed_result.text) @@ -1134,6 +1165,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return parsed_result @@ -1171,6 +1204,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_result except ValueError as e: @@ -1185,6 +1220,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return self._invoke_after_llm_call_hooks( @@ -1248,6 +1285,12 @@ class OpenAICompletion(BaseLLM): usage = self._extract_responses_token_usage(event.response) self._track_token_usage_internal(usage) + finish_reason, response_id = ( + self._extract_responses_finish_reason_and_id(final_response) + if final_response is not None + else (None, response_id_stream) + ) + if self.parse_tool_outputs and final_response: parsed_result = self._extract_builtin_tool_outputs(final_response) parsed_result.text = self._apply_stop_words(parsed_result.text) @@ -1259,6 +1302,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return parsed_result @@ -1296,6 +1341,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_result except ValueError as e: @@ -1310,6 +1357,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params.get("input", []), usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return full_response @@ -1603,6 +1652,9 @@ class OpenAICompletion(BaseLLM): usage = self._extract_openai_token_usage(parsed_response) self._track_token_usage_internal(usage) + parsed_finish_reason, parsed_response_id = ( + self._extract_chat_finish_reason_and_id(parsed_response) + ) parsed_object = parsed_response.choices[0].message.parsed if parsed_object: self._emit_call_completed_event( @@ -1612,6 +1664,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=parsed_finish_reason, + response_id=parsed_response_id, ) return parsed_object @@ -1625,6 +1679,9 @@ class OpenAICompletion(BaseLLM): choice: Choice = response.choices[0] message = choice.message + finish_reason, response_id = self._extract_chat_finish_reason_and_id( + response + ) # Without available_functions, return tool_calls so the caller (executor) handles execution if message.tool_calls and not available_functions: @@ -1635,6 +1692,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return list(message.tool_calls) @@ -1675,6 +1734,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_result except ValueError as e: @@ -1689,6 +1750,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) if usage.get("total_tokens", 0) > 0: @@ -1734,6 +1797,8 @@ class OpenAICompletion(BaseLLM): available_functions: dict[str, Any] | None = None, from_task: Any | None = None, from_agent: Any | None = None, + finish_reason: str | None = None, + response_id: str | None = None, ) -> str | list[dict[str, Any]]: """Finalize a streaming response with usage tracking, tool call handling, and events. @@ -1745,6 +1810,9 @@ class OpenAICompletion(BaseLLM): available_functions: Available functions for tool calling. from_task: Task that initiated the call. from_agent: Agent that initiated the call. + finish_reason: Raw provider finish reason (e.g. "stop", "length", + "tool_calls") extracted from the last streaming chunk. + response_id: Raw provider response id from any chunk. Returns: Tool calls list when tools were invoked without available_functions, @@ -1774,6 +1842,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) return tool_calls_list @@ -1817,6 +1887,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_data, + finish_reason=finish_reason, + response_id=response_id, ) return full_response @@ -1861,6 +1933,9 @@ class OpenAICompletion(BaseLLM): if final_completion: usage = self._extract_openai_token_usage(final_completion) self._track_token_usage_internal(usage) + parsed_finish_reason, parsed_response_id = ( + self._extract_chat_finish_reason_and_id(final_completion) + ) if final_completion.choices: parsed_result = final_completion.choices[0].message.parsed if parsed_result: @@ -1871,6 +1946,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=parsed_finish_reason, + response_id=parsed_response_id, ) return parsed_result @@ -1882,11 +1959,15 @@ class OpenAICompletion(BaseLLM): ) usage_data: dict[str, Any] | None = None + stream_finish_reason: str | None = None + stream_response_id: str | None = None for completion_chunk in completion_stream: response_id_stream = ( completion_chunk.id if hasattr(completion_chunk, "id") else None ) + if response_id_stream: + stream_response_id = response_id_stream if hasattr(completion_chunk, "usage") and completion_chunk.usage: usage_data = self._extract_openai_token_usage(completion_chunk) @@ -1897,6 +1978,9 @@ class OpenAICompletion(BaseLLM): choice = completion_chunk.choices[0] chunk_delta: ChoiceDelta = choice.delta + chunk_finish = getattr(choice, "finish_reason", None) + if chunk_finish: + stream_finish_reason = chunk_finish if chunk_delta.content: full_response += chunk_delta.content @@ -1954,6 +2038,8 @@ class OpenAICompletion(BaseLLM): available_functions=available_functions, from_task=from_task, from_agent=from_agent, + finish_reason=stream_finish_reason, + response_id=stream_response_id, ) if isinstance(result, str): return self._invoke_after_llm_call_hooks( @@ -1989,6 +2075,9 @@ class OpenAICompletion(BaseLLM): usage = self._extract_openai_token_usage(parsed_response) self._track_token_usage_internal(usage) + parsed_finish_reason, parsed_response_id = ( + self._extract_chat_finish_reason_and_id(parsed_response) + ) parsed_object = parsed_response.choices[0].message.parsed if parsed_object: self._emit_call_completed_event( @@ -1998,6 +2087,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=parsed_finish_reason, + response_id=parsed_response_id, ) return parsed_object @@ -2011,6 +2102,9 @@ class OpenAICompletion(BaseLLM): choice: Choice = response.choices[0] message = choice.message + finish_reason, response_id = self._extract_chat_finish_reason_and_id( + response + ) # Without available_functions, return tool_calls so the caller (executor) handles execution if message.tool_calls and not available_functions: @@ -2021,6 +2115,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return list(message.tool_calls) @@ -2065,6 +2161,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) return structured_result except ValueError as e: @@ -2079,6 +2177,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage, + finish_reason=finish_reason, + response_id=response_id, ) if usage.get("total_tokens", 0) > 0: @@ -2130,8 +2230,12 @@ class OpenAICompletion(BaseLLM): accumulated_content = "" usage_data: dict[str, Any] | None = None + parsed_stream_finish_reason: str | None = None + parsed_stream_response_id: str | None = None async for chunk in completion_stream: response_id_stream = chunk.id if hasattr(chunk, "id") else None + if response_id_stream: + parsed_stream_response_id = response_id_stream if hasattr(chunk, "usage") and chunk.usage: usage_data = self._extract_openai_token_usage(chunk) @@ -2142,6 +2246,9 @@ class OpenAICompletion(BaseLLM): choice = chunk.choices[0] delta: ChoiceDelta = choice.delta + chunk_finish = getattr(choice, "finish_reason", None) + if chunk_finish: + parsed_stream_finish_reason = chunk_finish if delta.content: accumulated_content += delta.content @@ -2165,6 +2272,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_data, + finish_reason=parsed_stream_finish_reason, + response_id=parsed_stream_response_id, ) return parsed_object @@ -2177,6 +2286,8 @@ class OpenAICompletion(BaseLLM): from_agent=from_agent, messages=params["messages"], usage=usage_data, + finish_reason=parsed_stream_finish_reason, + response_id=parsed_stream_response_id, ) return accumulated_content @@ -2185,9 +2296,13 @@ class OpenAICompletion(BaseLLM): ] = await self._get_async_client().chat.completions.create(**params) usage_data = None + stream_finish_reason: str | None = None + stream_response_id: str | None = None async for chunk in stream: response_id_stream = chunk.id if hasattr(chunk, "id") else None + if response_id_stream: + stream_response_id = response_id_stream if hasattr(chunk, "usage") and chunk.usage: usage_data = self._extract_openai_token_usage(chunk) @@ -2198,6 +2313,9 @@ class OpenAICompletion(BaseLLM): choice = chunk.choices[0] chunk_delta: ChoiceDelta = choice.delta + chunk_finish = getattr(choice, "finish_reason", None) + if chunk_finish: + stream_finish_reason = chunk_finish if chunk_delta.content: full_response += chunk_delta.content @@ -2255,6 +2373,8 @@ class OpenAICompletion(BaseLLM): available_functions=available_functions, from_task=from_task, from_agent=from_agent, + finish_reason=stream_finish_reason, + response_id=stream_response_id, ) def supports_function_calling(self) -> bool: @@ -2305,6 +2425,32 @@ class OpenAICompletion(BaseLLM): return int(8192 * CONTEXT_WINDOW_USAGE_RATIO) + def _effective_max_tokens(self) -> int | float | None: + """Newer OpenAI chat models cap via ``max_completion_tokens``.""" + return self.max_tokens or self.max_completion_tokens + + @staticmethod + def _extract_chat_finish_reason_and_id( + response: Any, + ) -> tuple[str | None, str | None]: + """ChatCompletion / ChatCompletionChunk share the choices-shape; + delegate to the shared extractor. + """ + return extract_choices_finish_reason_and_id(response) + + @staticmethod + def _extract_responses_finish_reason_and_id( + response: Any, + ) -> tuple[str | None, str | None]: + """Extract finish_reason and response_id from an OpenAI Responses + API ``Response`` object. The Responses API exposes ``status`` rather + than ``finish_reason``; we forward the raw status value. + """ + return ( + getattr(response, "status", None), + getattr(response, "id", None), + ) + def _extract_openai_token_usage( self, response: ChatCompletion | ChatCompletionChunk ) -> dict[str, Any]: diff --git a/lib/crewai/tests/events/test_llm_finish_reason_response_id.py b/lib/crewai/tests/events/test_llm_finish_reason_response_id.py new file mode 100644 index 000000000..091875fdf --- /dev/null +++ b/lib/crewai/tests/events/test_llm_finish_reason_response_id.py @@ -0,0 +1,526 @@ +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from crewai.events.event_bus import CrewAIEventsBus +from crewai.events.types.llm_events import ( + LLMCallCompletedEvent, + LLMCallStartedEvent, + LLMCallType, + LLMStreamChunkEvent, +) +from crewai.llm import LLM +from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id +from crewai.llms.base_llm import BaseLLM + + +class _StubLLM(BaseLLM): + model: str = "test-model" + + def call(self, *args: Any, **kwargs: Any) -> str: + return "" + + async def acall(self, *args: Any, **kwargs: Any) -> str: + return "" + + def supports_function_calling(self) -> bool: + return False + + +@pytest.fixture +def mock_emit(): + with patch.object(CrewAIEventsBus, "emit") as mock: + yield mock + + +class TestLLMCallCompletedEventFinishReasonAndResponseId: + def test_accepts_string_values(self): + event = LLMCallCompletedEvent( + response="hi", + call_type=LLMCallType.LLM_CALL, + call_id="call-1", + finish_reason="stop", + response_id="resp_123", + ) + assert event.finish_reason == "stop" + assert event.response_id == "resp_123" + + def test_defaults_to_none(self): + event = LLMCallCompletedEvent( + response="hi", + call_type=LLMCallType.LLM_CALL, + call_id="call-1", + ) + assert event.finish_reason is None + assert event.response_id is None + + @pytest.mark.parametrize( + "value", + [MagicMock(), 42, 1.5, ["stop"], {"reason": "stop"}, object()], + ) + def test_coerces_non_string_to_none(self, value): + event = LLMCallCompletedEvent( + response="hi", + call_type=LLMCallType.LLM_CALL, + call_id="call-1", + finish_reason=value, + response_id=value, + ) + assert event.finish_reason is None + assert event.response_id is None + + +class TestLLMCallStartedEventSamplingParams: + def test_accepts_all_sampling_params(self): + event = LLMCallStartedEvent( + call_id="call-1", + temperature=0.7, + top_p=0.9, + max_tokens=512, + stream=True, + seed=42, + stop_sequences=["END"], + frequency_penalty=0.1, + presence_penalty=0.2, + n=3, + ) + assert event.temperature == 0.7 + assert event.top_p == 0.9 + assert event.max_tokens == 512 + assert event.stream is True + assert event.seed == 42 + assert event.stop_sequences == ["END"] + assert event.frequency_penalty == 0.1 + assert event.presence_penalty == 0.2 + assert event.n == 3 + + def test_all_sampling_params_default_to_none(self): + event = LLMCallStartedEvent(call_id="call-1") + assert event.temperature is None + assert event.top_p is None + assert event.max_tokens is None + assert event.stream is None + assert event.seed is None + assert event.stop_sequences is None + assert event.frequency_penalty is None + assert event.presence_penalty is None + assert event.n is None + + +class TestStopSequencesCoercion: + # The OTel SDK falls back to str(value) when a span attribute isn't a + # recognised Sequence[str], producing the protobuf textproto repr + # ("values { string_value: ... }") in downstream telemetry. The + # field_validator coerces exotic iterables (Vertex/Gemini protobuf + # containers, tuples, generators) to a clean list[str] up front so the + # OTel attribute is always shaped correctly. + def test_bare_string_is_wrapped_in_list(self): + event = LLMCallStartedEvent(call_id="call-1", stop_sequences="\nObservation:") + assert event.stop_sequences == ["\nObservation:"] + + @pytest.mark.parametrize( + "raw, expected", + [ + (["\nObservation:", "Final Answer:"], ["\nObservation:", "Final Answer:"]), + (("\nObservation:",), ["\nObservation:"]), + ((s for s in ["a", "b"]), ["a", "b"]), + ([], []), + ], + ) + def test_python_iterables_pass_through( + self, raw: Any, expected: list[str] + ) -> None: + event = LLMCallStartedEvent(call_id="call-1", stop_sequences=raw) + assert event.stop_sequences == expected + + def test_protobuf_like_repeated_container_is_coerced(self): + # Mirrors google.protobuf RepeatedScalarContainer: iterable yielding + # actual Python str objects. Should pass through cleanly. + class _RepeatedScalar: + def __init__(self, items: list[str]) -> None: + self._items = items + + def __iter__(self): + return iter(self._items) + + event = LLMCallStartedEvent( + call_id="call-1", + stop_sequences=_RepeatedScalar(["\nObservation:"]), + ) + assert event.stop_sequences == ["\nObservation:"] + + def test_protobuf_listvalue_with_nested_values_coerces_to_textproto_strings(self): + # Mirrors google.protobuf.struct_pb2.ListValue: iterable yielding + # `Value` messages whose str() is "string_value: \"...\"". The + # coercion will str() each element, which is still wrong-shaped but + # at least lands as a real list[str] for the OTel attribute instead + # of a single textproto-blob string. Documents observed behaviour; + # the upstream fix is to pass list[str] to LLM.stop, not ListValue. + class _PbValue: + def __init__(self, string_value: str) -> None: + self.string_value = string_value + + def __str__(self) -> str: + return f'string_value: "{self.string_value}"' + + class _PbListValue: + def __init__(self, values: list[_PbValue]) -> None: + self.values = values + + def __iter__(self): + return iter(self.values) + + event = LLMCallStartedEvent( + call_id="call-1", + stop_sequences=_PbListValue([_PbValue("\\nObservation:")]), + ) + assert event.stop_sequences == ['string_value: "\\nObservation:"'] + + @pytest.mark.parametrize("bad_input", [123, 12.5, object()]) + def test_non_iterable_falls_back_to_none(self, bad_input: Any) -> None: + event = LLMCallStartedEvent(call_id="call-1", stop_sequences=bad_input) + assert event.stop_sequences is None + + def test_none_stays_none(self): + event = LLMCallStartedEvent(call_id="call-1", stop_sequences=None) + assert event.stop_sequences is None + + +class TestEmitCallStartedEventIntrospectsSamplingParams: + def test_reads_sampling_params_off_self(self, mock_emit): + llm = _StubLLM(model="test-model", temperature=0.4) + llm.top_p = 0.8 + llm.max_tokens = 256 + llm.stream = False + llm.seed = 7 + llm.frequency_penalty = 0.5 + llm.presence_penalty = 0.6 + llm.n = 2 + llm.stop = ["STOP"] + + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert isinstance(event, LLMCallStartedEvent) + assert event.temperature == 0.4 + assert event.top_p == 0.8 + assert event.max_tokens == 256 + assert event.stream is False + assert event.seed == 7 + assert event.stop_sequences == ["STOP"] + assert event.frequency_penalty == 0.5 + assert event.presence_penalty == 0.6 + assert event.n == 2 + + def test_explicit_kwargs_override_introspection(self, mock_emit): + llm = _StubLLM(model="test-model", temperature=0.4) + + llm._emit_call_started_event(messages="hi", temperature=0.9) + + event = mock_emit.call_args[1]["event"] + assert event.temperature == 0.9 + + +class TestBaseLLMSamplingParamFields: + # Regression: PR #5945 review feedback. Sampling params are declared as + # typed fields on BaseLLM so ``_emit_call_started_event`` reads them via + # plain attribute access instead of getattr/hasattr fallbacks. Kwargs + # like ``n=1`` bind directly to the typed field via Pydantic; there is + # no promotion from ``additional_params``. + def test_sampling_kwargs_bind_to_typed_fields(self, mock_emit): + from crewai.llms.providers.openai.completion import OpenAICompletion + + llm = LLM(model="gpt-4", n=1, temperature=0.5, seed=42) + + assert isinstance(llm, OpenAICompletion) + assert llm.n == 1 + assert llm.temperature == 0.5 + assert llm.seed == 42 + assert "n" not in llm.additional_params + assert "temperature" not in llm.additional_params + assert "seed" not in llm.additional_params + + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert isinstance(event, LLMCallStartedEvent) + assert event.n == 1 + assert event.temperature == 0.5 + assert event.seed == 42 + + def test_additional_params_are_not_promoted_to_typed_fields(self, mock_emit): + # Callers who pass sampling params through ``additional_params`` + # opt out of typed-field semantics. We intentionally do NOT promote + # those values back into ``self.n`` / ``self.temperature``, so the + # emitter sees ``None`` for those attributes. If a caller wants the + # value surfaced in telemetry, they pass it as a kwarg. + llm = LLM( + model="gpt-4", + additional_params={"n": 1, "temperature": 0.5, "seed": 42}, + ) + + assert llm.n is None + assert llm.temperature is None + assert llm.seed is None + assert llm.additional_params == {"n": 1, "temperature": 0.5, "seed": 42} + + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert isinstance(event, LLMCallStartedEvent) + assert event.n is None + assert event.temperature is None + assert event.seed is None + + def test_emit_uses_call_scoped_stop_override(self, mock_emit): + from crewai.llms.base_llm import call_stop_override + + llm = _StubLLM(model="test-model", stop=["A"]) + + with call_stop_override(llm, ["X"]): + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert isinstance(event, LLMCallStartedEvent) + assert event.stop_sequences == ["X"] + # Instance-level stop is never mutated by the override. + assert llm.stop == ["A"] + + +class TestEffectiveMaxTokensTelemetry: + def test_base_defaults_to_max_tokens(self, mock_emit): + llm = _StubLLM(model="test-model", max_tokens=256) + + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert event.max_tokens == 256 + + def test_openai_surfaces_max_completion_tokens(self, mock_emit): + from crewai.llms.providers.openai.completion import OpenAICompletion + + llm = LLM(model="gpt-4o", max_completion_tokens=512) + assert isinstance(llm, OpenAICompletion) + assert llm.max_tokens is None + + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert event.max_tokens == 512 + + def test_explicit_max_tokens_takes_precedence(self, mock_emit): + llm = LLM(model="gpt-4o", max_tokens=128, max_completion_tokens=512) + + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert event.max_tokens == 128 + + +class TestStreamingDictChunkResponseIdPropagation: + # Regression: PR #5945 coderabbitai feedback. The streaming loop only + # extracted ``chunk.id`` for ``ModelResponseBase`` instances; dict-shaped + # chunks (LiteLLM emits these in some configs) silently dropped the id + # and ``LLMStreamChunkEvent.response_id`` came through as ``None``. + def _dict_chunks(self) -> list[dict[str, Any]]: + return [ + { + "id": "test-chunk-id", + "choices": [{"delta": {"content": "hi"}, "finish_reason": None}], + }, + { + "id": "test-chunk-id", + "choices": [{"delta": {"content": " there"}, "finish_reason": "stop"}], + }, + ] + + def _stream_event_response_ids(self, mock_emit) -> list[str | None]: + return [ + call.kwargs["event"].response_id + for call in mock_emit.call_args_list + if isinstance(call.kwargs.get("event"), LLMStreamChunkEvent) + ] + + def test_sync_dict_chunk_id_propagates_to_stream_event(self, mock_emit): + llm = LLM(model="gpt-4o-mini", is_litellm=True, stream=True) + + with patch( + "crewai.llm.litellm.completion", + return_value=iter(self._dict_chunks()), + ): + llm.call("anything") + + ids = self._stream_event_response_ids(mock_emit) + assert ids, "expected at least one LLMStreamChunkEvent" + assert all(rid == "test-chunk-id" for rid in ids), ids + + @pytest.mark.asyncio + async def test_async_dict_chunk_id_propagates_to_stream_event(self, mock_emit): + llm = LLM(model="gpt-4o-mini", is_litellm=True, stream=True) + + async def _aiter(): + for chunk in self._dict_chunks(): + yield chunk + + async def _acompletion(*_args, **_kwargs): + return _aiter() + + with patch("crewai.llm.litellm.acompletion", side_effect=_acompletion): + await llm.acall("anything") + + ids = self._stream_event_response_ids(mock_emit) + assert ids, "expected at least one LLMStreamChunkEvent" + assert all(rid == "test-chunk-id" for rid in ids), ids + + +class TestEmitCallCompletedEventPassesFinishReasonAndResponseId: + def test_passes_through_to_event(self, mock_emit): + llm = _StubLLM(model="test-model") + + llm._emit_call_completed_event( + response="hi", + call_type=LLMCallType.LLM_CALL, + finish_reason="stop", + response_id="resp_123", + ) + + event = mock_emit.call_args[1]["event"] + assert isinstance(event, LLMCallCompletedEvent) + assert event.finish_reason == "stop" + assert event.response_id == "resp_123" + + def test_omitted_defaults_to_none(self, mock_emit): + llm = _StubLLM(model="test-model") + + llm._emit_call_completed_event( + response="hi", + call_type=LLMCallType.LLM_CALL, + ) + + event = mock_emit.call_args[1]["event"] + assert event.finish_reason is None + assert event.response_id is None + + +class TestLLMExtractFinishReasonAndResponseId: + def test_non_streaming_litellm_shape(self): + response = SimpleNamespace( + id="chatcmpl-abc", + choices=[SimpleNamespace(finish_reason="stop", message=SimpleNamespace())], + ) + + finish_reason, response_id = LLM._extract_finish_reason_and_response_id( + response + ) + + assert finish_reason == "stop" + assert response_id == "chatcmpl-abc" + + def test_streaming_litellm_chunk_shape(self): + last_chunk = SimpleNamespace( + id="chatcmpl-stream-xyz", + choices=[SimpleNamespace(finish_reason="tool_calls", delta=SimpleNamespace())], + ) + + finish_reason, response_id = LLM._extract_finish_reason_and_response_id( + last_chunk + ) + + assert finish_reason == "tool_calls" + assert response_id == "chatcmpl-stream-xyz" + + def test_dict_shape(self): + chunk = { + "id": "chatcmpl-dict", + "choices": [{"finish_reason": "length", "delta": {}}], + } + + finish_reason, response_id = LLM._extract_finish_reason_and_response_id(chunk) + + assert finish_reason == "length" + assert response_id == "chatcmpl-dict" + + def test_missing_fields_return_none(self): + finish_reason, response_id = LLM._extract_finish_reason_and_response_id( + SimpleNamespace() + ) + + assert finish_reason is None + assert response_id is None + + def test_non_string_values_coerced_to_none(self): + response = SimpleNamespace( + id=12345, + choices=[SimpleNamespace(finish_reason=MagicMock(), delta=SimpleNamespace())], + ) + + finish_reason, response_id = LLM._extract_finish_reason_and_response_id( + response + ) + + assert finish_reason is None + assert response_id is None + + def test_never_raises_on_unexpected_input(self): + assert LLM._extract_finish_reason_and_response_id(None) == (None, None) + assert LLM._extract_finish_reason_and_response_id(42) == (None, None) + assert LLM._extract_finish_reason_and_response_id("string") == (None, None) + + +class TestExtractChoicesFinishReasonAndIdHelper: + # The shared extractor is consumed by LLM (LiteLLM), OpenAI Chat, and Azure. + # TestLLMExtractFinishReasonAndResponseId exercises the choices-shape paths + # transitively; these tests cover the direct-call surface and the + # import contract. + @pytest.mark.parametrize( + "response, expected", + [ + ( + SimpleNamespace( + id="resp-1", choices=[SimpleNamespace(finish_reason="stop")] + ), + ("stop", "resp-1"), + ), + ( + {"id": "resp-2", "choices": [{"finish_reason": "length"}]}, + ("length", "resp-2"), + ), + ( + SimpleNamespace( + id="resp-3", choices=[{"finish_reason": "tool_calls"}] + ), + ("tool_calls", "resp-3"), + ), + ( + { + "id": "resp-4", + "choices": [SimpleNamespace(finish_reason="content_filter")], + }, + ("content_filter", "resp-4"), + ), + ], + ) + def test_extracts_choices_shape( + self, response: Any, expected: tuple[str | None, str | None] + ) -> None: + assert extract_choices_finish_reason_and_id(response) == expected + + @pytest.mark.parametrize( + "bad_input", + [ + None, + 42, + "string", + {}, + SimpleNamespace(), + SimpleNamespace(choices=[]), + SimpleNamespace(choices=[SimpleNamespace()]), + {"id": 12345, "choices": [{"finish_reason": MagicMock()}]}, + ], + ) + def test_never_raises_returns_nones_or_coerces(self, bad_input: Any) -> None: + finish_reason, response_id = extract_choices_finish_reason_and_id(bad_input) + assert finish_reason is None or isinstance(finish_reason, str) + assert response_id is None or isinstance(response_id, str) diff --git a/lib/crewai/tests/llms/google/test_google.py b/lib/crewai/tests/llms/google/test_google.py index 3bcdb0951..0213eb525 100644 --- a/lib/crewai/tests/llms/google/test_google.py +++ b/lib/crewai/tests/llms/google/test_google.py @@ -122,6 +122,20 @@ def test_gemini_completion_initialization_parameters(): assert llm.top_k == 40 +def test_gemini_started_event_surfaces_max_output_tokens(): + from crewai.events.event_bus import CrewAIEventsBus + from crewai.events.types.llm_events import LLMCallStartedEvent + + llm = LLM(model="google/gemini-2.0-flash-001", max_output_tokens=2000, api_key="test-key") + + with patch.object(CrewAIEventsBus, "emit") as mock_emit: + llm._emit_call_started_event(messages="hi") + + event = mock_emit.call_args[1]["event"] + assert isinstance(event, LLMCallStartedEvent) + assert event.max_tokens == 2000 + + def test_gemini_specific_parameters(): """ Test Gemini-specific parameters like stop_sequences, streaming, and safety settings diff --git a/lib/crewai/tests/test_llm_streaming_finish_reason.py b/lib/crewai/tests/test_llm_streaming_finish_reason.py new file mode 100644 index 000000000..ff8a94d4e --- /dev/null +++ b/lib/crewai/tests/test_llm_streaming_finish_reason.py @@ -0,0 +1,96 @@ +"""Regression: LiteLLM emits a final usage-only chunk (choices=[]) when +``stream_options.include_usage`` is set. The old post-loop +``_extract_finish_reason_and_response_id(last_chunk)`` then silently returned +(None, None). These tests pin that we capture finish_reason/response_id +incrementally during the stream loop instead. +""" +from __future__ import annotations + +from typing import Any +from unittest.mock import patch + +import pytest + +from crewai.events.event_bus import CrewAIEventsBus +from crewai.events.types.llm_events import LLMCallCompletedEvent +from crewai.llm import LLM + + +@pytest.fixture +def mock_emit(): + with patch.object(CrewAIEventsBus, "emit") as mock: + yield mock + + +def _completed_event(mock_emit) -> LLMCallCompletedEvent: + matches = [ + call.kwargs["event"] + for call in mock_emit.call_args_list + if isinstance(call.kwargs.get("event"), LLMCallCompletedEvent) + ] + assert matches, "expected an LLMCallCompletedEvent to be emitted" + assert len(matches) == 1, f"expected one completed event, got {len(matches)}" + return matches[0] + + +def _chunks_with_usage_tail() -> list[dict[str, Any]]: + """Three-chunk stream mirroring LiteLLM's include_usage behavior: + two content chunks where the second carries finish_reason="stop", + then a final usage-only chunk with choices=[].""" + return [ + { + "id": "chatcmpl-stream-1", + "choices": [ + {"delta": {"content": "hi"}, "finish_reason": None} + ], + }, + { + "id": "chatcmpl-stream-1", + "choices": [ + {"delta": {"content": " there"}, "finish_reason": "stop"} + ], + }, + { + "id": "chatcmpl-stream-1", + "choices": [], + "usage": { + "prompt_tokens": 1, + "completion_tokens": 2, + "total_tokens": 3, + }, + }, + ] + + +def test_sync_stream_emits_finish_reason_and_response_id_from_loop(mock_emit): + llm = LLM(model="gpt-4o-mini", is_litellm=True, stream=True) + + with patch("crewai.llm.litellm.completion", return_value=iter(_chunks_with_usage_tail())): + result = llm.call("anything") + + assert result == "hi there" + + event = _completed_event(mock_emit) + assert event.finish_reason == "stop" + assert event.response_id == "chatcmpl-stream-1" + + +@pytest.mark.asyncio +async def test_async_stream_emits_finish_reason_and_response_id_from_loop(mock_emit): + llm = LLM(model="gpt-4o-mini", is_litellm=True, stream=True) + + async def _aiter(): + for chunk in _chunks_with_usage_tail(): + yield chunk + + async def _acompletion(*_args, **_kwargs): + return _aiter() + + with patch("crewai.llm.litellm.acompletion", side_effect=_acompletion): + result = await llm.acall("anything") + + assert result == "hi there" + + event = _completed_event(mock_emit) + assert event.finish_reason == "stop" + assert event.response_id == "chatcmpl-stream-1"