Skip to content

refactor: Sync agent.py with llm.py patterns - Phase 1 #765

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jul 8, 2025
Merged

refactor: Sync agent.py with llm.py patterns - Phase 1 #765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
beff3d0
refactor: Sync agent.py with llm.py patterns - Phase 1
claude[bot] Jul 8, 2025
77a3030
Rename test_agent_refactor.py to src/praisonai/tests/unit/test_agent_…
MervinPraison Jul 8, 2025
debbd9c
fix: Replace shallow copy with deep copy in _build_messages()
claude[bot] Jul 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
230 changes: 130 additions & 100 deletions src/praisonai-agents/praisonaiagents/agent/agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import os
import time
import json
import copy
import logging
import asyncio
from typing import List, Optional, Any, Dict, Union, Literal, TYPE_CHECKING, Callable, Tuple
Expand Down Expand Up @@ -831,6 +832,127 @@ def _apply_guardrail_with_retry(self, response_text, prompt, temperature=0.2, to

return current_response

def _build_messages(self, prompt, temperature=0.2, output_json=None, output_pydantic=None):
"""Build messages list for chat completion.

Args:
prompt: The user prompt (str or list)
temperature: Temperature for the chat
output_json: Optional Pydantic model for JSON output
output_pydantic: Optional Pydantic model for JSON output (alias)

Returns:
tuple: (messages list, original prompt)
"""
messages = []

# Build system prompt if enabled
system_prompt = None
if self.use_system_prompt:
system_prompt = f"""{self.backstory}\n
Your Role: {self.role}\n
Your Goal: {self.goal}
"""
if output_json:
system_prompt += f"\nReturn ONLY a JSON object that matches this Pydantic model: {json.dumps(output_json.model_json_schema())}"
elif output_pydantic:
system_prompt += f"\nReturn ONLY a JSON object that matches this Pydantic model: {json.dumps(output_pydantic.model_json_schema())}"

messages.append({"role": "system", "content": system_prompt})

# Add chat history
messages.extend(self.chat_history)

# Handle prompt modifications for JSON output
original_prompt = prompt
if output_json or output_pydantic:
if isinstance(prompt, str):
prompt = prompt + "\nReturn ONLY a valid JSON object. No other text or explanation."
elif isinstance(prompt, list):
# Create a deep copy to avoid modifying the original
prompt = copy.deepcopy(prompt)
for item in prompt:
if item.get("type") == "text":
item["text"] = item["text"] + "\nReturn ONLY a valid JSON object. No other text or explanation."
break

# Add prompt to messages
if isinstance(prompt, list):
# If we receive a multimodal prompt list, place it directly in the user message
messages.append({"role": "user", "content": prompt})
else:
messages.append({"role": "user", "content": prompt})

return messages, original_prompt

def _format_tools_for_completion(self, tools=None):
"""Format tools for OpenAI completion API.

Supports:
- Pre-formatted OpenAI tools (dicts with type='function')
- Lists of pre-formatted tools
- Callable functions
- String function names
- Objects with to_openai_tool() method

Args:
tools: List of tools in various formats or None to use self.tools

Returns:
List of formatted tools or empty list
"""
if tools is None:
tools = self.tools

if not tools:
return []

formatted_tools = []
for tool in tools:
# Handle pre-formatted OpenAI tools
if isinstance(tool, dict) and tool.get('type') == 'function':
# Validate nested dictionary structure before accessing
if 'function' in tool and isinstance(tool['function'], dict) and 'name' in tool['function']:
formatted_tools.append(tool)
else:
logging.warning(f"Skipping malformed OpenAI tool: missing function or name")
# Handle lists of tools
elif isinstance(tool, list):
for subtool in tool:
if isinstance(subtool, dict) and subtool.get('type') == 'function':
# Validate nested dictionary structure before accessing
if 'function' in subtool and isinstance(subtool['function'], dict) and 'name' in subtool['function']:
formatted_tools.append(subtool)
else:
logging.warning(f"Skipping malformed OpenAI tool in list: missing function or name")
# Handle string tool names
elif isinstance(tool, str):
tool_def = self._generate_tool_definition(tool)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Could not generate definition for tool: {tool}")
# Handle objects with to_openai_tool method (MCP tools)
elif hasattr(tool, "to_openai_tool"):
formatted_tools.append(tool.to_openai_tool())
# Handle callable functions
elif callable(tool):
tool_def = self._generate_tool_definition(tool.__name__)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Tool {tool} not recognized")

# Validate JSON serialization before returning
if formatted_tools:
try:
json.dumps(formatted_tools) # Validate serialization
except (TypeError, ValueError) as e:
logging.error(f"Tools are not JSON serializable: {e}")
return []

return formatted_tools
Comment on lines +888 to +954
Copy link
Contributor

@coderabbitai coderabbitai bot Jul 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Fix f-string issues and improve error handling consistency

The tool formatting consolidation is well-implemented but has a couple of issues to address:

  1. F-string issues: Lines 918 and 927 use f-strings without placeholders
  2. Inconsistent error handling: Malformed tools generate warnings but processing continues, while JSON serialization errors cause the entire tool list to be discarded

Apply this diff to fix the f-string issues:

-                    logging.warning(f"Skipping malformed OpenAI tool: missing function or name")
+                    logging.warning("Skipping malformed OpenAI tool: missing function or name")
                else:
-                        logging.warning(f"Skipping malformed OpenAI tool in list: missing function or name")
+                        logging.warning("Skipping malformed OpenAI tool in list: missing function or name")

Consider making error handling more consistent by either:

  • Skipping individual malformed tools for both validation and serialization errors
  • Or failing the entire operation uniformly for all error types

Add type hints for better IDE support:

-def _format_tools_for_completion(self, tools=None):
+def _format_tools_for_completion(self, tools: Optional[List[Any]] = None) -> List[Dict]:
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def _format_tools_for_completion(self, tools=None):
"""Format tools for OpenAI completion API.
Supports:
- Pre-formatted OpenAI tools (dicts with type='function')
- Lists of pre-formatted tools
- Callable functions
- String function names
- Objects with to_openai_tool() method
Args:
tools: List of tools in various formats or None to use self.tools
Returns:
List of formatted tools or empty list
"""
if tools is None:
tools = self.tools
if not tools:
return []
formatted_tools = []
for tool in tools:
# Handle pre-formatted OpenAI tools
if isinstance(tool, dict) and tool.get('type') == 'function':
# Validate nested dictionary structure before accessing
if 'function' in tool and isinstance(tool['function'], dict) and 'name' in tool['function']:
formatted_tools.append(tool)
else:
logging.warning(f"Skipping malformed OpenAI tool: missing function or name")
# Handle lists of tools
elif isinstance(tool, list):
for subtool in tool:
if isinstance(subtool, dict) and subtool.get('type') == 'function':
# Validate nested dictionary structure before accessing
if 'function' in subtool and isinstance(subtool['function'], dict) and 'name' in subtool['function']:
formatted_tools.append(subtool)
else:
logging.warning(f"Skipping malformed OpenAI tool in list: missing function or name")
# Handle string tool names
elif isinstance(tool, str):
tool_def = self._generate_tool_definition(tool)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Could not generate definition for tool: {tool}")
# Handle objects with to_openai_tool method (MCP tools)
elif hasattr(tool, "to_openai_tool"):
formatted_tools.append(tool.to_openai_tool())
# Handle callable functions
elif callable(tool):
tool_def = self._generate_tool_definition(tool.__name__)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Tool {tool} not recognized")
# Validate JSON serialization before returning
if formatted_tools:
try:
json.dumps(formatted_tools) # Validate serialization
except (TypeError, ValueError) as e:
logging.error(f"Tools are not JSON serializable: {e}")
return []
return formatted_tools
def _format_tools_for_completion(self, tools: Optional[List[Any]] = None) -> List[Dict]:
"""Format tools for OpenAI completion API.
Supports:
- Pre-formatted OpenAI tools (dicts with type='function')
- Lists of pre-formatted tools
- Callable functions
- String function names
- Objects with to_openai_tool() method
Args:
tools: List of tools in various formats or None to use self.tools
Returns:
List of formatted tools or empty list
"""
if tools is None:
tools = self.tools
if not tools:
return []
formatted_tools = []
for tool in tools:
# Handle pre-formatted OpenAI tools
if isinstance(tool, dict) and tool.get('type') == 'function':
# Validate nested dictionary structure before accessing
if 'function' in tool and isinstance(tool['function'], dict) and 'name' in tool['function']:
formatted_tools.append(tool)
else:
logging.warning("Skipping malformed OpenAI tool: missing function or name")
# Handle lists of tools
elif isinstance(tool, list):
for subtool in tool:
if isinstance(subtool, dict) and subtool.get('type') == 'function':
# Validate nested dictionary structure before accessing
if 'function' in subtool and isinstance(subtool['function'], dict) and 'name' in subtool['function']:
formatted_tools.append(subtool)
else:
logging.warning("Skipping malformed OpenAI tool in list: missing function or name")
# Handle string tool names
elif isinstance(tool, str):
tool_def = self._generate_tool_definition(tool)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Could not generate definition for tool: {tool}")
# Handle objects with to_openai_tool method (MCP tools)
elif hasattr(tool, "to_openai_tool"):
formatted_tools.append(tool.to_openai_tool())
# Handle callable functions
elif callable(tool):
tool_def = self._generate_tool_definition(tool.__name__)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Tool {tool} not recognized")
# Validate JSON serialization before returning
if formatted_tools:
try:
json.dumps(formatted_tools) # Validate serialization
except (TypeError, ValueError) as e:
logging.error(f"Tools are not JSON serializable: {e}")
return []
return formatted_tools
🧰 Tools
🪛 Ruff (0.11.9)

918-918: f-string without any placeholders

Remove extraneous f prefix

(F541)

927-927: f-string without any placeholders

Remove extraneous f prefix

(F541)

🤖 Prompt for AI Agents 
In src/praisonai-agents/praisonaiagents/agent/agent.py lines 888 to 954, fix the
f-string issues by removing unnecessary f-string syntax on lines 918 and 927
where no placeholders are used, replacing them with regular string literals.
Improve error handling consistency by ensuring that malformed tools detected
during validation or JSON serialization are handled uniformly—either skip only
the problematic tools without discarding the entire list or fail the entire
operation consistently. Additionally, add appropriate type hints to the
_format_tools_for_completion method signature and relevant variables to enhance
IDE support and code clarity.


def generate_task(self) -> 'Task':
"""Generate a Task object from the agent's instructions"""
from ..task.task import Task
Expand Down Expand Up @@ -1045,26 +1167,8 @@ def _chat_completion(self, messages, temperature=0.2, tools=None, stream=True, r
start_time = time.time()
logging.debug(f"{self.name} sending messages to LLM: {messages}")

formatted_tools = []
if tools is None:
tools = self.tools
if tools:
for tool in tools:
if isinstance(tool, str):
# Generate tool definition for string tool names
tool_def = self._generate_tool_definition(tool)
if tool_def:
formatted_tools.append(tool_def)
else:
logging.warning(f"Could not generate definition for tool: {tool}")
elif isinstance(tool, dict):
formatted_tools.append(tool)
elif hasattr(tool, "to_openai_tool"):
formatted_tools.append(tool.to_openai_tool())
elif callable(tool):
formatted_tools.append(self._generate_tool_definition(tool.__name__))
else:
logging.warning(f"Tool {tool} not recognized")
# Use the new _format_tools_for_completion helper method
formatted_tools = self._format_tools_for_completion(tools)

try:
# Use the custom LLM instance if available
Expand Down Expand Up @@ -1297,40 +1401,8 @@ def chat(self, prompt, temperature=0.2, tools=None, output_json=None, output_pyd
display_error(f"Error in LLM chat: {e}")
return None
else:
if self.use_system_prompt:
system_prompt = f"""{self.backstory}\n
Your Role: {self.role}\n
Your Goal: {self.goal}
"""
if output_json:
system_prompt += f"\nReturn ONLY a JSON object that matches this Pydantic model: {json.dumps(output_json.model_json_schema())}"
elif output_pydantic:
system_prompt += f"\nReturn ONLY a JSON object that matches this Pydantic model: {json.dumps(output_pydantic.model_json_schema())}"
else:
system_prompt = None

messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.extend(self.chat_history)

# Modify prompt if output_json or output_pydantic is specified
original_prompt = prompt
if output_json or output_pydantic:
if isinstance(prompt, str):
prompt += "\nReturn ONLY a valid JSON object. No other text or explanation."
elif isinstance(prompt, list):
# For multimodal prompts, append to the text content
for item in prompt:
if item["type"] == "text":
item["text"] += "\nReturn ONLY a valid JSON object. No other text or explanation."
break

if isinstance(prompt, list):
# If we receive a multimodal prompt list, place it directly in the user message
messages.append({"role": "user", "content": prompt})
else:
messages.append({"role": "user", "content": prompt})
# Use the new _build_messages helper method
messages, original_prompt = self._build_messages(prompt, temperature, output_json, output_pydantic)

final_response_text = None
reflection_count = 0
Expand Down Expand Up @@ -1566,38 +1638,8 @@ async def achat(self, prompt: str, temperature=0.2, tools=None, output_json=None
return None

# For OpenAI client
if self.use_system_prompt:
system_prompt = f"""{self.backstory}\n
Your Role: {self.role}\n
Your Goal: {self.goal}
"""
if output_json:
system_prompt += f"\nReturn ONLY a JSON object that matches this Pydantic model: {json.dumps(output_json.model_json_schema())}"
elif output_pydantic:
system_prompt += f"\nReturn ONLY a JSON object that matches this Pydantic model: {json.dumps(output_pydantic.model_json_schema())}"
else:
system_prompt = None

messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.extend(self.chat_history)

# Modify prompt if output_json or output_pydantic is specified
original_prompt = prompt
if output_json or output_pydantic:
if isinstance(prompt, str):
prompt += "\nReturn ONLY a valid JSON object. No other text or explanation."
elif isinstance(prompt, list):
for item in prompt:
if item["type"] == "text":
item["text"] += "\nReturn ONLY a valid JSON object. No other text or explanation."
break

if isinstance(prompt, list):
messages.append({"role": "user", "content": prompt})
else:
messages.append({"role": "user", "content": prompt})
# Use the new _build_messages helper method
messages, original_prompt = self._build_messages(prompt, temperature, output_json, output_pydantic)

reflection_count = 0
start_time = time.time()
Expand All @@ -1619,20 +1661,8 @@ async def achat(self, prompt: str, temperature=0.2, tools=None, output_json=None
agent_tools=agent_tools
)

# Format tools if provided
formatted_tools = []
if tools:
for tool in tools:
if isinstance(tool, str):
tool_def = self._generate_tool_definition(tool)
if tool_def:
formatted_tools.append(tool_def)
elif isinstance(tool, dict):
formatted_tools.append(tool)
elif hasattr(tool, "to_openai_tool"):
formatted_tools.append(tool.to_openai_tool())
elif callable(tool):
formatted_tools.append(self._generate_tool_definition(tool.__name__))
# Use the new _format_tools_for_completion helper method
formatted_tools = self._format_tools_for_completion(tools)

# Create async OpenAI client
async_client = AsyncOpenAI()
Expand Down
Loading
Loading