Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# object_oriented_agents/core_classes/agent_signature.py

from typing import Optional, Dict, Any, List
from .tool_manager import ToolManager

Expand All @@ -8,30 +9,36 @@ class AgentSignature:
- The developer prompt
- The model name
- The list of tool definitions
- The default reasoning effort (if any)
"""

def __init__(self, developer_prompt: str, model_name: str, tool_manager: Optional[ToolManager] = None):
def __init__(self, developer_prompt: str, model_name: str, tool_manager: Optional[ToolManager] = None, reasoning_effort: Optional[str] = None):
self.developer_prompt = developer_prompt
self.model_name = model_name
self.tool_manager = tool_manager
self.reasoning_effort = reasoning_effort

def to_dict(self) -> Dict[str, Any]:
"""
Return a dictionary containing:
1. The developer prompt
2. The model name
3. A list of tool definitions (function schemas)
4. The default reasoning effort if defined
"""
if self.tool_manager:
# Each item in get_tool_definitions() looks like {"type": "function", "function": {...}}
tool_definitions = self.tool_manager.get_tool_definitions()
# We need the whole definition for the final signature
functions = [t for t in tool_definitions]
else:
functions = []

return {
signature_dict = {
"developer_prompt": self.developer_prompt,
"model_name": self.model_name,
"tools": functions
}
}
if self.reasoning_effort is not None:
signature_dict["reasoning_effort"] = self.reasoning_effort

return signature_dict
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# object_oriented_agents/core_classes/base_agent.py

from abc import ABC, abstractmethod
from typing import Optional
from .chat_messages import ChatMessages
Expand All @@ -19,14 +20,16 @@ def __init__(
developer_prompt: str,
model_name: str,
logger=None,
language_model_interface: LanguageModelInterface = None
language_model_interface: LanguageModelInterface = None,
reasoning_effort: Optional[str] = None
):
self.developer_prompt = developer_prompt
self.model_name = model_name
self.messages = ChatMessages(developer_prompt)
self.tool_manager: Optional[ToolManager] = None
self.logger = logger or get_logger(self.__class__.__name__)
self.language_model_interface = language_model_interface
self.reasoning_effort = reasoning_effort

@abstractmethod
def setup_tools(self) -> None:
Expand All @@ -40,12 +43,16 @@ def add_message(self, content: str) -> None:
self.logger.debug(f"Adding user message: {content}")
self.messages.add_user_message(content)

def task(self, user_task: str, tool_call_enabled: bool = True, return_tool_response_as_is: bool = False) -> str:
def task(self, user_task: str, tool_call_enabled: bool = True, return_tool_response_as_is: bool = False,
reasoning_effort: Optional[str] = None) -> str:
# Use the reasoning_effort provided in the method call if present, otherwise fall back to the agent's default
final_reasoning_effort = reasoning_effort if reasoning_effort is not None else self.reasoning_effort

if self.language_model_interface is None:
error_message = "Error: Cannot execute task without the LanguageModelInterface."
self.logger.error(error_message)
raise ValueError(error_message)

self.logger.debug(f"Starting task: {user_task} (tool_call_enabled={tool_call_enabled})")

# Add user message
Expand All @@ -56,13 +63,17 @@ def task(self, user_task: str, tool_call_enabled: bool = True, return_tool_respo
tools = self.tool_manager.get_tool_definitions()
self.logger.debug(f"Tools available: {tools}")

# Submit to OpenAI
# Build parameter dict and include reasoning_effort only if not None
params = {
"model": self.model_name,
"messages": self.messages.get_messages(),
"tools": tools
}
if final_reasoning_effort is not None:
params["reasoning_effort"] = final_reasoning_effort

self.logger.debug("Sending request to language model interface...")
response = self.language_model_interface.generate_completion(
model=self.model_name,
messages=self.messages.get_messages(),
tools=tools,
)
response = self.language_model_interface.generate_completion(**params)

tool_calls = response.choices[0].message.tool_calls
if tool_call_enabled and self.tool_manager and tool_calls:
Expand All @@ -71,7 +82,8 @@ def task(self, user_task: str, tool_call_enabled: bool = True, return_tool_respo
response,
return_tool_response_as_is,
self.messages,
self.model_name
self.model_name,
reasoning_effort=final_reasoning_effort
)

# No tool call, normal assistant response
Expand All @@ -86,10 +98,12 @@ def signature(self) -> dict:
- The developer prompt
- The model name
- The tool definitions (function schemas)
- The default reasoning effort if set
"""
signature_obj = AgentSignature(
developer_prompt=self.developer_prompt,
model_name=self.model_name,
tool_manager=self.tool_manager
tool_manager=self.tool_manager,
reasoning_effort=self.reasoning_effort
)
return signature_obj.to_dict()
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
# object_oriented_agents/core_classes/tool_manager.py

import json
from typing import Dict, Any, List
from typing import Dict, Any, List, Optional
from .chat_messages import ChatMessages
from .tool_interface import ToolInterface
from ..utils.logger import get_logger
Expand All @@ -16,7 +16,6 @@ class ToolManager:
- Handle the entire tool call sequence
"""


def __init__(self, logger=None, language_model_interface: LanguageModelInterface = None):
self.tools = {}
self.logger = logger or get_logger(self.__class__.__name__)
Expand Down Expand Up @@ -47,7 +46,8 @@ def handle_tool_call_sequence(
response,
return_tool_response_as_is: bool,
messages: ChatMessages,
model_name: str
model_name: str,
reasoning_effort: Optional[str] = None
) -> str:
"""
If the model wants to call a tool, parse the function arguments, invoke the tool,
Expand Down Expand Up @@ -90,11 +90,15 @@ def handle_tool_call_sequence(
complete_payload.append(function_call_result_message)

self.logger.debug("Calling the model again with the tool response to get the final answer.")
# Use the injected openai_client here
response_after_tool_call = self.language_model_interface.generate_completion(
model=model_name,
messages=complete_payload
)
# Build parameter dict and only include reasoning_effort if not None
params = {
"model": model_name,
"messages": complete_payload
}
if reasoning_effort is not None:
params["reasoning_effort"] = reasoning_effort

response_after_tool_call = self.language_model_interface.generate_completion(**params)

final_message = response_after_tool_call.choices[0].message.content
self.logger.debug("Received final answer from model after tool call.")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,17 @@ def generate_completion(
self,
model: str,
messages: List[Dict[str, str]],
tools: Optional[List[Dict[str, Any]]] = None
tools: Optional[List[Dict[str, Any]]] = None,
reasoning_effort: Optional[str] = None
) -> Dict[str, Any]:
"""
Generate a completion (response) from the language model given a set of messages and optional tool definitions.
Generate a completion (response) from the language model given a set of messages, optional tool definitions,
and an optional reasoning effort parameter.

:param model: The name of the model to call.
:param messages: A list of messages, where each message is a dict with keys 'role' and 'content'.
:param tools: Optional list of tool definitions.
:param reasoning_effort: Optional parameter to indicate additional reasoning effort.
:return: A dictionary representing the model's response. The shape of this dict follows the provider's format.
"""
pass
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,11 @@ def generate_completion(
self,
model: str,
messages: List[Dict[str, str]],
tools: Optional[List[Dict[str, Any]]] = None
tools: Optional[List[Dict[str, Any]]] = None,
reasoning_effort: Optional[str] = None
) -> Dict[str, Any]:
"""
Calls the OpenAI API to generate a chat completion using the provided messages and tools.
Calls the OpenAI API to generate a chat completion using the provided messages, tools, and optional reasoning_effort.
"""
kwargs = {
"model": model,
Expand All @@ -34,6 +35,10 @@ def generate_completion(
# Adjust this as necessary if the API format changes.
kwargs["tools"] = tools

# Append reasoning_effort to kwargs if provided
if reasoning_effort is not None:
kwargs["reasoning_effort"] = reasoning_effort

self.logger.debug("Generating completion with OpenAI model.")
self.logger.debug(f"Request: {kwargs}")
try:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
# Create a LanguageModelInterface instance using the OpenAILanguageModel
language_model_api_interface = OpenAILanguageModel(api_key=os.getenv("OPENAI_API_KEY"), logger=myapp_logger)


class PythonExecAgent(BaseAgent):
"""
An agent specialized in executing Python code in a Docker container.
Expand All @@ -33,17 +34,18 @@ def __init__(
3. Generate Python code to analyze the data and call the tool `execute_python_code` to run the code and get results.
4. You can use Python libraries pandas, numpy, matplotlib, seaborn, and scikit-learn.
5. Interpret the results of the code execution and provide analysis to the user.

""",
model_name: str = "o3-mini",
logger=myapp_logger,
language_model_interface = language_model_api_interface
):
language_model_interface=language_model_api_interface,
reasoning_effort: str = None # optional; if provided, passed to API calls
):
super().__init__(
developer_prompt=developer_prompt,
model_name=model_name,
logger=logger,
language_model_interface=language_model_interface
language_model_interface=language_model_interface,
reasoning_effort=reasoning_effort
)
self.setup_tools()

Expand All @@ -52,9 +54,9 @@ def setup_tools(self) -> None:
Create a ToolManager, instantiate the PythonExecTool and register it with the ToolManager.
"""
self.tool_manager = ToolManager(logger=self.logger, language_model_interface=self.language_model_interface)

# Create the Python execution tool
python_exec_tool = PythonExecTool()

# Register the Python execution tool
self.tool_manager.register_tool(python_exec_tool)
4 changes: 2 additions & 2 deletions registry.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -1798,9 +1798,9 @@
- usage-api
- cost-api

- title: Build Your Own Code Interpreter - Empowering LLM Agents with Dynamic Tool Calling
- title: Build Your Own Code Interpreter - Dynamic Tool Generation and Execution With o3-mini
path: examples/object_oriented_agentic_approach/Secure_code_interpreter_tool_for_LLM_agents.ipynb
date: 2025-01-26
date: 2025-02-03
authors:
- msingh-openai
tags:
Expand Down