diff --git a/src/cai/cli.py b/src/cai/cli.py index 164352f7..9eb97ba1 100644 --- a/src/cai/cli.py +++ b/src/cai/cli.py @@ -128,7 +128,10 @@ from cai.agents import get_agent_by_name # to preserve conversation context between turns. from cai.sdk.agents.models.openai_chatcompletions import ( message_history, + add_to_message_history, ) +from cai.sdk.agents.items import ToolCallOutputItem +from cai.sdk.agents.stream_events import RunItemStreamEvent # Load environment variables from .env file load_dotenv() @@ -387,8 +390,33 @@ def run_cai_cli(starting_agent, context_variables=None, stream=False, max_turns= for msg in message_history: role = msg.get("role") content = msg.get("content") - if role in ("user", "assistant") and content: - history_context.append({"role": role, "content": content}) + tool_calls = msg.get("tool_calls") + + if role == "user": + history_context.append({"role": "user", "content": content or ""}) + elif role == "system": + history_context.append({"role": "system", "content": content or ""}) + elif role == "assistant": + if tool_calls: + history_context.append( + { + "role": "assistant", + "content": content, # Can be None + "tool_calls": tool_calls, + } + ) + elif content is not None: + history_context.append({"role": "assistant", "content": content}) + elif content is None and not tool_calls: # Explicitly handle empty assistant message + history_context.append({"role": "assistant", "content": None}) + elif role == "tool": + history_context.append( + { + "role": "tool", + "tool_call_id": msg.get("tool_call_id"), + "content": msg.get("content"), # Tool output + } + ) # Append the current user input as the last message in the list. conversation_input: list | str @@ -405,8 +433,17 @@ def run_cai_cli(starting_agent, context_variables=None, stream=False, max_turns= result = Runner.run_streamed(agent, conversation_input) # Consume events so the async generator is executed. - async for _ in result.stream_events(): - pass + async for event in result.stream_events(): + if isinstance(event, RunItemStreamEvent) and event.name == "tool_output": + # Ensure item is a ToolCallOutputItem before accessing attributes + if isinstance(event.item, ToolCallOutputItem): + tool_msg = { + "role": "tool", + "tool_call_id": event.item.raw_item["call_id"], # Changed to dictionary access + "content": event.item.output, + } + add_to_message_history(tool_msg) + # pass # Original logic was just pass return result except Exception as e: @@ -422,6 +459,14 @@ def run_cai_cli(starting_agent, context_variables=None, stream=False, max_turns= else: # Use non-streamed response response = asyncio.run(Runner.run(agent, conversation_input)) + for item in response.new_items: + if isinstance(item, ToolCallOutputItem): + tool_msg = { + "role": "tool", + "tool_call_id": item.raw_item["call_id"], # Usar formato consistente con streaming + "content": item.output, + } + add_to_message_history(tool_msg) turn_count += 1 # Stop measuring active time and start measuring idle time again diff --git a/src/cai/sdk/agents/models/openai_chatcompletions.py b/src/cai/sdk/agents/models/openai_chatcompletions.py index 5eb36c90..932c218d 100644 --- a/src/cai/sdk/agents/models/openai_chatcompletions.py +++ b/src/cai/sdk/agents/models/openai_chatcompletions.py @@ -137,7 +137,6 @@ def add_to_message_history(msg): existing.get("content") == msg.get("content") for existing in message_history ) - elif msg.get("role") == "assistant" and msg.get("tool_calls"): is_duplicate = any( existing.get("role") == "assistant" and @@ -145,6 +144,12 @@ def add_to_message_history(msg): existing["tool_calls"][0].get("id") == msg["tool_calls"][0].get("id") for existing in message_history ) + elif msg.get("role") == "tool": + is_duplicate = any( + existing.get("role") == "tool" and + existing.get("tool_call_id") == msg.get("tool_call_id") + for existing in message_history + ) if not is_duplicate: message_history.append(msg) @@ -534,6 +539,37 @@ class OpenAIChatCompletionsModel(Model): """ Yields a partial message as it is generated, as well as the usage information. """ + # IMPORTANT: Pre-process input to ensure it's in the correct format + # for streaming. This helps prevent errors during stream handling. + if not isinstance(input, str): + # Convert input items to messages and verify structure + try: + input_items = list(input) # Make sure it's a list + # Pre-verify the input messages to avoid errors during streaming + from cai.util import fix_message_list + + # Apply fix_message_list to the input items that are dictionaries + dict_items = [item for item in input_items if isinstance(item, dict)] + if dict_items: + fixed_dict_items = fix_message_list(dict_items) + + # Replace the original dict items with fixed ones while preserving non-dict items + new_input = [] + dict_index = 0 + for item in input_items: + if isinstance(item, dict): + if dict_index < len(fixed_dict_items): + new_input.append(fixed_dict_items[dict_index]) + dict_index += 1 + else: + new_input.append(item) + + # Update input with the fixed version + input = new_input + except Exception as e: + print(f"Warning: Error pre-processing input for streaming: {e}") + # Continue with original input even if pre-processing failed + # Increment the interaction counter for CLI display self.interaction_counter += 1 @@ -1299,6 +1335,13 @@ class OpenAIChatCompletionsModel(Model): if tracing.include_data(): span.span_data.input = converted_messages + # Ensure message list has correct structure regardless of error condition + 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}") + parallel_tool_calls = ( True if model_settings.parallel_tool_calls and tools and len(tools) > 0 else NOT_GIVEN ) @@ -1481,13 +1524,19 @@ class OpenAIChatCompletionsModel(Model): return await self._fetch_response_litellm_ollama(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) elif ("An assistant message with 'tool_calls'" in str(e) or - "`tool_use` blocks must be followed by a user message with `tool_result`" in str(e)): # noqa: E501 # pylint: disable=C0301 + "`tool_use` blocks must be followed by a user message with `tool_result`" in str(e) or # noqa: E501 # pylint: disable=C0301 + "An assistant message with 'tool_calls' must be followed by tool messages" in str(e)): # Añadir esta condición print(f"Error: {str(e)}") # NOTE: EDGE CASE: Report Agent CTRL C error # # This fix CTRL-C error when message list is incomplete # When a tool is not finished but the LLM generates a tool call - kwargs["messages"] = fix_message_list(kwargs["messages"]) + try: + from cai.util import fix_message_list + kwargs["messages"] = fix_message_list(kwargs["messages"]) + except Exception as fix_error: + print(f"Failed to fix message sequence: {fix_error}") + return await self._fetch_response_litellm_openai(kwargs, model_settings, tool_choice, stream, parallel_tool_calls) # this captures an error related to the fact @@ -1989,6 +2038,47 @@ class _Converter: return current_assistant_msg for item in items: + # NEW: Handle 'tool' messages from history + if ( + isinstance(item, dict) + and item.get("role") == "tool" + and "tool_call_id" in item + and "content" in item + ): + flush_assistant_message() # Ensure any pending assistant message is flushed + tool_message: ChatCompletionToolMessageParam = { + "role": "tool", + "tool_call_id": item["tool_call_id"], + "content": str(item["content"] or ""), # Ensure content is a string + } + result.append(tool_message) + continue + + # 0) Assistant messages with tool_calls only (from memory) + if ( + isinstance(item, dict) + and item.get("role") == "assistant" + and item.get("tool_calls") + ): + flush_assistant_message() + tool_calls_param: list[ChatCompletionMessageToolCallParam] = [] + for tc in item["tool_calls"]: + tool_calls_param.append( + ChatCompletionMessageToolCallParam( + id=tc.get("id", ""), + type=tc.get("type", "function"), + function=tc.get("function", {}), + ) + ) + msg_asst: ChatCompletionAssistantMessageParam = { + "role": "assistant", + "content": item.get("content"), + "tool_calls": tool_calls_param, + } + result.append(msg_asst) + # Skip further processing for this item + continue + # 1) Check easy input message if easy_msg := cls.maybe_easy_input_message(item): role = easy_msg["role"] diff --git a/src/cai/util.py b/src/cai/util.py index 1555159f..cc6fc0c7 100644 --- a/src/cai/util.py +++ b/src/cai/util.py @@ -549,91 +549,76 @@ def fix_message_list(messages): # pylint: disable=R0914,R0915,R0912 List[dict]: Sanitized list of messages with invalid tool calls and empty messages removed. """ - # Step 1: Filter and discard empty messages (considered empty if 'content' - # is None or only whitespace) - cleaned_messages = [] - for msg in messages: - content = msg.get("content") - if content is not None and content.strip(): - cleaned_messages.append(msg) - messages = cleaned_messages - # Step 2: Collect tool call id occurrences. - # In assistant messages, iterate through 'tool_calls' list. - # In 'tool' type messages, use the 'tool_call_id' key. - tool_calls_occurrences = {} - for i, msg in enumerate(messages): - if msg.get("role") == "assistant" and isinstance( - msg.get("tool_calls"), list): - for j, tool_call in enumerate(msg["tool_calls"]): - tc_id = tool_call.get("id") - if tc_id: - tool_calls_occurrences.setdefault( - tc_id, []).append((i, "assistant", j)) - elif msg.get("role") == "tool" and msg.get("tool_call_id"): - tc_id = msg["tool_call_id"] - tool_calls_occurrences.setdefault( - tc_id, []).append((i, "tool", None)) - - # Step 3: Mark indices in the message list to remove. - # Maps message index (assistant) to set of indices (in tool_calls) to - # delete, or directly marks message indices (tool) to delete. - to_remove = {} - for tc_id, occurrences in tool_calls_occurrences.items(): - if len(occurrences) > 2: - # More than one assistant and tool message pair - trim down - # by picking first pairing and removing the rest - assistant_items = [ - occ for occ in occurrences if occ[1] == "assistant"] - tool_items = [occ for occ in occurrences if occ[1] == "tool"] - - if assistant_items and tool_items: - valid_assistant = assistant_items[0] - valid_tool = tool_items[0] - for item in occurrences: - if item != valid_assistant and item != valid_tool: - if item[1] == "assistant": - # If assistant message, mark specific tool_call index - to_remove.setdefault(item[0], set()).add(item[2]) - else: - # If tool message, mark whole message - to_remove[item[0]] = None - else: - # Only one type of message, no complete pairs - remove them all - for item in occurrences: - if item[1] == "assistant": - to_remove.setdefault(item[0], set()).add(item[2]) - else: - to_remove[item[0]] = None - elif len(occurrences) == 1: - # Incomplete pair (only tool call without tool result or vice versa) - item = occurrences[0] - if item[1] == "assistant": - to_remove.setdefault(item[0], set()).add(item[2]) - else: - to_remove[item[0]] = None - - # Step 4: Apply the removals and reconstruct the message list + # Deep-copy to ensure we don't modify the input sanitized_messages = [] + + # First pass - identify tool_call_ids from assistant messages and tool messages + tool_call_map = {} # Map from tool_call_id to (assistant_idx, tool_idx) + for i, msg in enumerate(messages): - if i in to_remove and to_remove[i] is None: - # Skip entirely removed messages + # Skip empty messages (considered empty if 'content' is None or only whitespace) + if msg.get("role") in ["user", "system"] and (msg.get("content") is None or not msg.get("content").strip()): continue - - # For assistant messages, remove marked tool_calls - if msg.get("role") == "assistant" and "tool_calls" in msg: - new_tool_calls = [] - for j, tc in enumerate(msg["tool_calls"]): - if i not in to_remove or j not in to_remove[i]: - new_tool_calls.append(tc) - msg["tool_calls"] = new_tool_calls - # If after modification message has no content and no tool_calls, - # skip it - if not (msg.get("content", "").strip() or - not msg.get("tool_calls")): - continue - + + # Add valid messages to our sanitized list first sanitized_messages.append(msg) - + + # Now track tool calls and tool messages for pairing + if msg.get("role") == "assistant" and msg.get("tool_calls"): + for tc in msg["tool_calls"]: + if tc.get("id"): + tool_id = tc.get("id") + if tool_id not in tool_call_map: + tool_call_map[tool_id] = {"assistant_idx": len(sanitized_messages) - 1, "tool_idx": None} + + if msg.get("role") == "tool" and msg.get("tool_call_id"): + tool_id = msg.get("tool_call_id") + if tool_id in tool_call_map: + tool_call_map[tool_id]["tool_idx"] = len(sanitized_messages) - 1 + else: + # Tool response without a matching tool call - create a synthetic pair + # by adding a dummy assistant message with a tool_call + assistant_msg = { + "role": "assistant", + "content": None, + "tool_calls": [{ + "id": tool_id, + "type": "function", + "function": { + "name": "unknown_function", + "arguments": "{}" + } + }] + } + # Insert the assistant message *before* the tool message + sanitized_messages.insert(len(sanitized_messages) - 1, assistant_msg) + # Update mapping + tool_call_map[tool_id] = {"assistant_idx": len(sanitized_messages) - 2, "tool_idx": len(sanitized_messages) - 1} + + # Final validation - ensure all tool calls have responses + for tool_id, indices in tool_call_map.items(): + if indices["tool_idx"] is None: + # Tool call without a response - create a synthetic tool message + assistant_idx = indices["assistant_idx"] + assistant_msg = sanitized_messages[assistant_idx] + + # Find the relevant tool call + tool_name = "unknown_function" + for tc in assistant_msg["tool_calls"]: + if tc.get("id") == tool_id: + if tc.get("function") and tc["function"].get("name"): + tool_name = tc["function"]["name"] + break + + # Create an automatic tool response message + tool_msg = { + "role": "tool", + "tool_call_id": tool_id, + "content": f"Auto-generated response for {tool_name}" + } + # Insert right after the assistant message + sanitized_messages.insert(assistant_idx + 1, tool_msg) + return sanitized_messages def cli_print_tool_call(tool_name="", args="", output="", prefix=" "):