diff --git a/src/cai/sdk/agents/models/openai_chatcompletions.py b/src/cai/sdk/agents/models/openai_chatcompletions.py index 0a5412ad..665c41d9 100644 --- a/src/cai/sdk/agents/models/openai_chatcompletions.py +++ b/src/cai/sdk/agents/models/openai_chatcompletions.py @@ -289,68 +289,44 @@ class OpenAIChatCompletionsModel(Model): stop_idle_timer() start_active_timer() + # Process current input (user messages, tool results from previous turn) + # and add them to the global message_history. + # _Converter.items_to_messages converts input to ChatCompletionMessageParam list + current_turn_chat_completion_params = _Converter.items_to_messages(input) + for msg_param in current_turn_chat_completion_params: + # Ensure these are plain dicts for add_to_message_history if they are typed objects + # add_to_message_history expects dicts. ChatCompletionMessageParam are TypedDicts. + add_to_message_history(cast(dict, msg_param)) + + # Ensure system instructions are in history (add_to_message_history handles duplicates) + if system_instructions: + sys_msg_for_history = {"role": "system", "content": system_instructions} + add_to_message_history(sys_msg_for_history) + + # Original logic for preparing converted_messages for token counting and logging (local scope) + # This specific `converted_messages` variable is for logging, the API call will use message_history. + # However, for consistent logging, it should reflect what's sent. + # Let's defer defining this until after _fetch_response returns the actual sent messages. + with generation_span( model=str(self.model), model_config=dataclasses.asdict(model_settings) | {"base_url": str(self._client.base_url)}, disabled=tracing.is_disabled(), ) as span_generation: - # Prepare the messages for consistent token counting - converted_messages = _Converter.items_to_messages(input) - if system_instructions: - converted_messages.insert( - 0, - { - "content": system_instructions, - "role": "system", - }, - ) - # --- Add to message_history: user, system, and assistant tool call messages --- - # Add system prompt to message_history - if system_instructions: - sys_msg = { - "role": "system", - "content": system_instructions - } - add_to_message_history(sys_msg) - - # Add user prompt(s) to message_history - if isinstance(input, str): - user_msg = { - "role": "user", - "content": input - } - add_to_message_history(user_msg) - # Log the user message - self.logger.log_user_message(input) - elif isinstance(input, list): - for item in input: - # Try to extract user messages - if isinstance(item, dict): - if item.get("role") == "user": - user_msg = { - "role": "user", - "content": item.get("content", "") - } - add_to_message_history(user_msg) - # Log the user message - if item.get("content"): - self.logger.log_user_message(item.get("content")) + # Get token count estimate using the current state of message_history for better accuracy + # as this reflects what will be sent to _fetch_response. + # Note: fix_message_list might alter it further, this is an estimate. + estimating_messages = list(message_history) # Use a copy + # Append current_turn_chat_completion_params for estimation if not already fully reflected + # This part is tricky as add_to_message_history might deduplicate. + # For simplicity, let's assume message_history is now the source for estimation. + estimated_input_tokens, _ = count_tokens_with_tiktoken(list(message_history)) - # IMPORTANT: Ensure the message list has valid tool call/result pairs - # This needs to happen before the API call to prevent errors - try: - from cai.util import fix_message_list - converted_messages = fix_message_list(converted_messages) - except Exception as e: - logger.warning(f"Failed to fix message list: {e}") - - # Get token count estimate before API call for consistent counting - estimated_input_tokens, _ = count_tokens_with_tiktoken(converted_messages) - - response = await self._fetch_response( - system_instructions, - input, + # _fetch_response will now use the global message_history + api_response, messages_sent_to_api = await self._fetch_response( + # system_instructions, # Removed + # input, # Removed model_settings, tools, output_schema, @@ -360,84 +336,79 @@ class OpenAIChatCompletionsModel(Model): stream=False, ) + # Use messages_sent_to_api for logging and further processing if needed + final_converted_messages_for_log = messages_sent_to_api + if _debug.DONT_LOG_MODEL_DATA: logger.debug("Received model response") else: logger.debug( - f"LLM resp:\n{json.dumps(response.choices[0].message.model_dump(), indent=2)}\n" + f"LLM resp:\\n{json.dumps(api_response.choices[0].message.model_dump(), indent=2)}\\n" ) # Ensure we have reasonable token counts - if response.usage: - input_tokens = response.usage.prompt_tokens - output_tokens = response.usage.completion_tokens - total_tokens = response.usage.total_tokens + if api_response.usage: + input_tokens = api_response.usage.prompt_tokens + output_tokens = api_response.usage.completion_tokens + # total_tokens = api_response.usage.total_tokens # total_tokens might be unused # Use estimated tokens if API returns zeroes or implausible values - if input_tokens == 0 or input_tokens < (len(str(input)) // 10): # Sanity check + # Compare against the length of the messages actually sent + if input_tokens == 0 or input_tokens < (len(json.dumps(messages_sent_to_api)) // 20): # Heuristic input_tokens = estimated_input_tokens - total_tokens = input_tokens + output_tokens - - # # Debug information - # print(f"\nDEBUG CONSISTENT TOKEN COUNTS - API tokens: input={input_tokens}, output={output_tokens}, total={total_tokens}") - # print(f"Estimated tokens were: input={estimated_input_tokens}") + # total_tokens = input_tokens + output_tokens else: # If no usage info, use our estimates input_tokens = estimated_input_tokens - output_tokens = 0 - total_tokens = input_tokens - # print(f"\nDEBUG CONSISTENT TOKEN COUNTS - No API tokens, using estimates: input={input_tokens}, output={output_tokens}") + output_tokens = 0 # Output tokens can't be estimated accurately before response + # total_tokens = input_tokens # Update token totals for CLI display self.total_input_tokens += input_tokens - self.total_output_tokens += output_tokens - if (response.usage and - hasattr(response.usage, 'completion_tokens_details') and - response.usage.completion_tokens_details and - hasattr(response.usage.completion_tokens_details, 'reasoning_tokens')): - self.total_reasoning_tokens += response.usage.completion_tokens_details.reasoning_tokens + self.total_output_tokens += output_tokens # This should be from actual response usage + if hasattr(api_response.usage, 'completion_tokens') and api_response.usage.completion_tokens is not None: + self.total_output_tokens = self.total_output_tokens - output_tokens + api_response.usage.completion_tokens # adjust if output_tokens was 0 + output_tokens = api_response.usage.completion_tokens + + + if (api_response.usage and + hasattr(api_response.usage, 'completion_tokens_details') and + api_response.usage.completion_tokens_details and + hasattr(api_response.usage.completion_tokens_details, 'reasoning_tokens')): + self.total_reasoning_tokens += api_response.usage.completion_tokens_details.reasoning_tokens # Check if this message contains tool calls - tool_output = None + # tool_output = None # tool_output seems unused here should_display_message = True - if (hasattr(response.choices[0].message, 'tool_calls') and - response.choices[0].message.tool_calls): + if (hasattr(api_response.choices[0].message, 'tool_calls') and + api_response.choices[0].message.tool_calls): - # For each tool call in the message, get corresponding output if available - for tool_call in response.choices[0].message.tool_calls: + for tool_call in api_response.choices[0].message.tool_calls: call_id = tool_call.id - - # If we're using direct tool output display with cli_print_tool_output, - # and we've already displayed this tool call output, we can skip displaying - # the assistant message to avoid duplication if (hasattr(_Converter, 'tool_outputs') and call_id in _Converter.tool_outputs and hasattr(_Converter, 'recent_tool_calls') and call_id in _Converter.recent_tool_calls): - # We've already displayed this tool and its output directly should_display_message = False break - # Only display the agent message if we haven't already shown the tool output if should_display_message: - # Ensure we're in non-streaming mode for proper markdown parsing previous_stream_setting = os.environ.get('CAI_STREAM', 'false') - os.environ['CAI_STREAM'] = 'false' # Force non-streaming mode for markdown parsing + os.environ['CAI_STREAM'] = 'false' - # Print the agent message for CLI display cli_print_agent_messages( agent_name=getattr(self, 'agent_name', 'Agent'), - message=response.choices[0].message, + message=api_response.choices[0].message, counter=getattr(self, 'interaction_counter', 0), model=str(self.model), debug=False, interaction_input_tokens=input_tokens, interaction_output_tokens=output_tokens, interaction_reasoning_tokens=( - response.usage.completion_tokens_details.reasoning_tokens - if response.usage and hasattr(response.usage, 'completion_tokens_details') - and response.usage.completion_tokens_details - and hasattr(response.usage.completion_tokens_details, 'reasoning_tokens') + api_response.usage.completion_tokens_details.reasoning_tokens + if api_response.usage and hasattr(api_response.usage, 'completion_tokens_details') + and api_response.usage.completion_tokens_details + and hasattr(api_response.usage.completion_tokens_details, 'reasoning_tokens') else 0 ), total_input_tokens=getattr(self, 'total_input_tokens', 0), @@ -445,118 +416,96 @@ class OpenAIChatCompletionsModel(Model): total_reasoning_tokens=getattr(self, 'total_reasoning_tokens', 0), interaction_cost=None, total_cost=None, - tool_output=None, # Don't pass tool output here, we're using direct display - suppress_empty=True # Suppress empty panels + tool_output=None, + suppress_empty=True ) - # Restore previous streaming setting os.environ['CAI_STREAM'] = previous_stream_setting - # --- Add assistant tool call to message_history if present --- - # If the response contains tool_calls, add them to message_history as assistant messages - assistant_msg = response.choices[0].message - if hasattr(assistant_msg, "tool_calls") and assistant_msg.tool_calls: - for tool_call in assistant_msg.tool_calls: - # Compose a message for the tool call - tool_call_msg = { + assistant_msg_from_api = api_response.choices[0].message + if hasattr(assistant_msg_from_api, "tool_calls") and assistant_msg_from_api.tool_calls: + for tool_call_param in assistant_msg_from_api.tool_calls: + tool_call_dict = { "role": "assistant", - "content": None, + "content": None, # Or assistant_msg_from_api.content if it can coexist "tool_calls": [ { - "id": tool_call.id, - "type": tool_call.type, + "id": tool_call_param.id, + "type": tool_call_param.type, "function": { - "name": tool_call.function.name, - "arguments": tool_call.function.arguments + "name": tool_call_param.function.name, + "arguments": tool_call_param.function.arguments } } ] } + add_to_message_history(tool_call_dict) - add_to_message_history(tool_call_msg) - - # Save the tool call details for later matching with output - # This is important for non-streaming mode to track tool calls properly if not hasattr(_Converter, 'recent_tool_calls'): _Converter.recent_tool_calls = {} - - # Store the tool call by ID for later reference - import time - _Converter.recent_tool_calls[tool_call.id] = { - 'name': tool_call.function.name, - 'arguments': tool_call.function.arguments, - 'start_time': time.time(), - 'execution_info': { - 'start_time': time.time() - } + _Converter.recent_tool_calls[tool_call_param.id] = { + 'name': tool_call_param.function.name, + 'arguments': tool_call_param.function.arguments, + 'start_time': time.time(), # This time might be slightly off; tool call already received + 'execution_info': {'start_time': time.time()} } - # Log the assistant tool call message - tool_calls_list = [] - for tool_call in assistant_msg.tool_calls: - tool_calls_list.append({ - "id": tool_call.id, - "type": tool_call.type, - "function": { - "name": tool_call.function.name, - "arguments": tool_call.function.arguments - } - }) - self.logger.log_assistant_message(None, tool_calls_list) - # If the assistant message is just text, add it as well - elif hasattr(assistant_msg, "content") and assistant_msg.content: - asst_msg = { + tool_calls_for_log = [{ + "id": tc.id, "type": tc.type, + "function": {"name": tc.function.name, "arguments": tc.function.arguments} + } for tc in assistant_msg_from_api.tool_calls] + self.logger.log_assistant_message(None, tool_calls_for_log) + + elif hasattr(assistant_msg_from_api, "content") and assistant_msg_from_api.content: + asst_msg_for_history = { "role": "assistant", - "content": assistant_msg.content + "content": assistant_msg_from_api.content } - add_to_message_history(asst_msg) - # Log the assistant message - self.logger.log_assistant_message(assistant_msg.content) + add_to_message_history(asst_msg_for_history) + self.logger.log_assistant_message(assistant_msg_from_api.content) - # Log the complete response for the session self.logger.rec_training_data( { "model": str(self.model), - "messages": converted_messages, + "messages": final_converted_messages_for_log, # Use the messages actually sent "stream": False, "tools": [t.params_json_schema for t in tools] if tools else [], "tool_choice": model_settings.tool_choice }, - response, + api_response, # This is ChatCompletion, not the Response object self.total_cost ) - usage = ( + usage_obj = ( Usage( requests=1, input_tokens=input_tokens, - output_tokens=output_tokens, + output_tokens=output_tokens, # Ensure this is the final output_tokens total_tokens=input_tokens + output_tokens, ) - if response.usage or input_tokens > 0 + if api_response.usage or input_tokens > 0 else Usage() ) if tracing.include_data(): - span_generation.span_data.output = [response.choices[0].message.model_dump()] + span_generation.span_data.output = [assistant_msg_from_api.model_dump()] span_generation.span_data.usage = { - "input_tokens": usage.input_tokens, - "output_tokens": usage.output_tokens, + "input_tokens": usage_obj.input_tokens, + "output_tokens": usage_obj.output_tokens, } - items = _Converter.message_to_output_items(response.choices[0].message) + items = _Converter.message_to_output_items(assistant_msg_from_api) - # For non-streaming responses, make sure we also log token usage with compatible field names - # This ensures both streaming and non-streaming use consistent naming - if not hasattr(response, 'usage'): - response.usage = {} - if hasattr(response.usage, 'prompt_tokens') and not hasattr(response.usage, 'input_tokens'): - response.usage.input_tokens = response.usage.prompt_tokens - if hasattr(response.usage, 'completion_tokens') and not hasattr(response.usage, 'output_tokens'): - response.usage.output_tokens = response.usage.completion_tokens + # Ensure usage compatibility for ModelResponse + final_usage = {} + if api_response.usage: + final_usage['input_tokens'] = api_response.usage.prompt_tokens + final_usage['output_tokens'] = api_response.usage.completion_tokens + final_usage['total_tokens'] = api_response.usage.total_tokens + # else: ModelResponse will use its default Usage() if no API usage return ModelResponse( output=items, - usage=usage, + usage=Usage(**final_usage) if final_usage else usage_obj, # Pass constructed Usage referenceable_id=None, ) @@ -1362,7 +1311,7 @@ class OpenAIChatCompletionsModel(Model): self.logger.rec_training_data( { "model": str(self.model), - "messages": converted_messages, + "messages": final_converted_messages_for_log, # Use the messages actually sent "stream": True, "tools": [t.params_json_schema for t in tools] if tools else [], "tool_choice": model_settings.tool_choice @@ -1392,8 +1341,8 @@ class OpenAIChatCompletionsModel(Model): @overload async def _fetch_response( self, - system_instructions: str | None, - input: str | list[TResponseInputItem], + # system_instructions: str | None, # REMOVED + # input: str | list[TResponseInputItem], # REMOVED model_settings: ModelSettings, tools: list[Tool], output_schema: AgentOutputSchema | None, @@ -1401,13 +1350,13 @@ class OpenAIChatCompletionsModel(Model): span: Span[GenerationSpanData], tracing: ModelTracing, stream: Literal[True], - ) -> tuple[Response, AsyncStream[ChatCompletionChunk]]: ... + ) -> tuple[Response, AsyncStream[ChatCompletionChunk], list[dict]]: ... # Added messages_sent_to_api @overload async def _fetch_response( self, - system_instructions: str | None, - input: str | list[TResponseInputItem], + # system_instructions: str | None, # REMOVED + # input: str | list[TResponseInputItem], # REMOVED model_settings: ModelSettings, tools: list[Tool], output_schema: AgentOutputSchema | None, @@ -1415,12 +1364,12 @@ class OpenAIChatCompletionsModel(Model): span: Span[GenerationSpanData], tracing: ModelTracing, stream: Literal[False], - ) -> ChatCompletion: ... + ) -> tuple[ChatCompletion, list[dict]]: ... # Added messages_sent_to_api async def _fetch_response( self, - system_instructions: str | None, - input: str | list[TResponseInputItem], + # system_instructions: str | None, # REMOVED + # input: str | list[TResponseInputItem], # REMOVED model_settings: ModelSettings, tools: list[Tool], output_schema: AgentOutputSchema | None, @@ -1428,38 +1377,37 @@ class OpenAIChatCompletionsModel(Model): span: Span[GenerationSpanData], tracing: ModelTracing, stream: bool = False, - ) -> ChatCompletion | tuple[Response, AsyncStream[ChatCompletionChunk]]: + ) -> ChatCompletion | tuple[Response, AsyncStream[ChatCompletionChunk]] | tuple[ChatCompletion, list[dict]] | tuple[Response, AsyncStream[ChatCompletionChunk], list[dict]]: # start by re-fetching self.is_ollama self.is_ollama = os.getenv('OLLAMA') is not None and os.getenv('OLLAMA').lower() == 'true' - converted_messages = _Converter.items_to_messages(input) + # Messages for API are now derived from the global message_history + messages_for_api = list(message_history) # Create a mutable copy - if system_instructions: - converted_messages.insert( - 0, - { - "content": system_instructions, - "role": "system", - }, - ) if tracing.include_data(): - span.span_data.input = converted_messages + span.span_data.input = messages_for_api # Log the full list - # IMPORTANT: Always sanitize the message list to prevent tool call errors - # This is critical to fix common errors with tool/assistant sequences + # IMPORTANT: Always sanitize the message list (which is the full history) try: from cai.util import fix_message_list - prev_length = len(converted_messages) - converted_messages = fix_message_list(converted_messages) - new_length = len(converted_messages) + prev_length = len(messages_for_api) + # Critical: fix_message_list operates on the full history now + messages_for_api_fixed = fix_message_list(messages_for_api) + new_length = len(messages_for_api_fixed) - # Log if the message list was changed significantly if new_length != prev_length: - logger.debug(f"Message list was fixed: {prev_length} -> {new_length} messages") + logger.debug(f"Message list was fixed: {prev_length} -> {new_length} messages. Old: {messages_for_api}, New: {messages_for_api_fixed}") + messages_for_api = messages_for_api_fixed # Use the fixed list except Exception as e: - logger.warning(f"Failed to fix message list: {e}") - + # Log more detailed error if fix_message_list fails + logger.error(f"CRITICAL: fix_message_list failed on message history: {e}", exc_info=True) + logger.error(f"Message history that caused failure: {json.dumps(messages_for_api, indent=2)}") + # It's crucial to understand why fix_message_list might fail here. + # Re-raise or handle as an AgentsException if appropriate. + raise AgentsException(f"Message history sanitization failed processing full history: {e}") from e + + # parallel_tool_calls, tool_choice, response_format, converted_tools... parallel_tool_calls = ( True if model_settings.parallel_tool_calls and tools and len(tools) > 0 else NOT_GIVEN ) @@ -1474,7 +1422,7 @@ class OpenAIChatCompletionsModel(Model): logger.debug("Calling LLM") else: logger.debug( - f"{json.dumps(converted_messages, indent=2)}\n" + f"{json.dumps(messages_for_api, indent=2)}\n" f"Tools:\n{json.dumps(converted_tools, indent=2)}\n" f"Stream: {stream}\n" f"Tool choice: {tool_choice}\n" @@ -1495,7 +1443,7 @@ class OpenAIChatCompletionsModel(Model): # Prepare kwargs for the API call kwargs = { "model": agent_model if agent_model else self.model, - "messages": converted_messages, + "messages": messages_for_api, # Use the fully prepared and fixed message list "tools": converted_tools or NOT_GIVEN, "temperature": self._non_null_or_not_given(model_settings.temperature), "top_p": self._non_null_or_not_given(model_settings.top_p), @@ -1579,9 +1527,21 @@ class OpenAIChatCompletionsModel(Model): try: if self.is_ollama: - return await self._fetch_response_litellm_ollama(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) + # Adjust Ollama fetch to return messages_for_api + if stream: + response_obj, stream_obj = await self._fetch_response_litellm_ollama(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) + return response_obj, stream_obj, messages_for_api + else: + completion = await self._fetch_response_litellm_ollama(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) + return completion, messages_for_api else: - return await self._fetch_response_litellm_openai(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) + # Adjust OpenAI fetch to return messages_for_api + if stream: + response_obj, stream_obj = await self._fetch_response_litellm_openai(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) + return response_obj, stream_obj, messages_for_api + else: + completion = await self._fetch_response_litellm_openai(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) + return completion, messages_for_api except litellm.exceptions.BadRequestError as e: # print(color("BadRequestError encountered: " + str(e), fg="yellow")) @@ -1770,11 +1730,11 @@ class OpenAIChatCompletionsModel(Model): tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven, stream: bool, parallel_tool_calls: bool - ) -> ChatCompletion | tuple[Response, AsyncStream[ChatCompletionChunk]]: + ) -> ChatCompletion | tuple[Response, AsyncStream[ChatCompletionChunk]]: # Return type will be wrapped by _fetch_response """Handle standard LiteLLM API calls for OpenAI and compatible models.""" if stream: # Standard LiteLLM handling for streaming - ret = litellm.completion(**kwargs) + # ret = litellm.completion(**kwargs) # This was likely a typo, should be acompletion for stream stream_obj = await litellm.acompletion(**kwargs) response = Response( @@ -1803,7 +1763,7 @@ class OpenAIChatCompletionsModel(Model): stream: bool, parallel_tool_calls: bool, provider="ollama" - ) -> ChatCompletion | tuple[Response, AsyncStream[ChatCompletionChunk]]: + ) -> ChatCompletion | tuple[Response, AsyncStream[ChatCompletionChunk]]: # Return type will be wrapped by _fetch_response # Extract only supported parameters for Ollama ollama_supported_params = { "model": kwargs.get("model", ""), diff --git a/src/cai/tools/common.py b/src/cai/tools/common.py index 2650a827..1846d56b 100644 --- a/src/cai/tools/common.py +++ b/src/cai/tools/common.py @@ -1249,7 +1249,7 @@ def run_command(command, ctf=None, stdout=False, # pylint: disable=too-many-arg command, stdout, timeout, - stream=stream, + stream=True, call_id=call_id, tool_name=tool_name, workspace_dir=_get_workspace_dir(),