RomanNum3ral
/
cai
mirror of https://github.com/aliasrobotics/cai.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'recurrent_messages' into '0.4.0' 

Improve message list handling and formating

See merge request aliasrobotics/alias_research/cai!160
This commit is contained in:
Luis Javier Navarrete Lozano 2025-05-08 18:36:54 +00:00
parent 21023df785 52b4cb3af5
commit a3639b69bf
3 changed files with 208 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
src/cai/cli.py
View File

@ -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

96
src/cai/sdk/agents/models/openai_chatcompletions.py
View File

@ -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"]

147
src/cai/util.py
View File

@ -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=" "):