Msingh agentic systems code interpreter (#1639)

Author: msingh-openai

examples/object_oriented_agentic_approach/Secure_code_interpreter_tool_for_LLM_agents.ipynb

@@ -0,0 +1,92 @@
+classDiagram
+
+    %% ==========================
+    %% Abstract Classes
+    %% ==========================
+    class BaseAgent {
+        - model_name: str
+        - messages: ChatMessages
+        - tool_manager: ToolManager
+        - logger: Logger
+        - language_model_interface: LanguageModelInterface
+
+        + __init__(developer_prompt: str, model_name: str)
+        + setup_tools(): void  %% abstract
+        + add_context(content: str): void
+        + add_message(content: str): void
+        + task(user_task: str, tool_call_enabled: bool, return_tool_response_as_is: bool): str
+    }
+
+    class ToolInterface {
+        + get_definition(): Dict[str, Any]
+        + run(arguments: Dict[str, Any]): str
+    }
+
+    class LanguageModelInterface {
+        + generate_completion(
+            model: str, 
+            messages: List<Dict[str, str]], 
+            tools: List[Dict[str, Any]]
+        ): Dict[str, Any]
+    }
+
+
+    %% ==========================
+    %% Concrete Classes
+    %% ==========================
+    class ChatMessages {
+        - messages: List<Dict[str, str]>
+
+        + __init__(developer_prompt: str)
+        + add_developer_message(content: str): void
+        + add_user_message(content: str): void
+        + add_assistant_message(content: str): void
+        + get_messages(): List<Dict[str, str]>
+    }
+
+    class ToolManager {
+        - tools: Dict[str, ToolInterface]
+        - logger: Logger
+        - language_model_interface: LanguageModelInterface
+
+        + __init__()
+        + register_tool(tool: ToolInterface): void
+        + get_tool_definitions(): List<Dict[str, Any]>
+        + handle_tool_call_sequence(
+            response: str, 
+            return_tool_response_as_is: bool, 
+            messages: ChatMessages
+        ): str
+    }
+
+    class OpenAILanguageModel {
+        - openai_client: OpenAI
+        - logger: Logger
+
+        + __init__()
+        + generate_completion(
+            model: str, 
+            messages: List<Dict[str, str]], 
+            tools: List<Dict[str, Any]]
+        ): Dict[str, Any]
+    }
+
+    class OpenAIClientFactory {
+        + create_client(api_key: str): OpenAI
+        - _resolve_api_key(api_key: str): str
+    }
+
+
+    %% ==========================
+    %% Relationships
+    %% ==========================
+    BaseAgent --> ChatMessages : uses
+    BaseAgent --> LanguageModelInterface : uses
+    BaseAgent --> ToolManager : optionally uses
+
+    ToolManager "0..*" --> "1" ToolInterface : manages
+    ToolManager --> ChatMessages : uses
+    ToolManager --> LanguageModelInterface : uses
+
+    LanguageModelInterface <|-- OpenAILanguageModel
+    OpenAILanguageModel --> OpenAIClientFactory : uses

examples/object_oriented_agentic_approach/resources/data/traffic_accidents.csv

@@ -0,0 +1,15 @@
+FROM python:3.10
+
+RUN apt-get update && \
+    apt-get install -y build-essential && \
+    rm -rf /var/lib/apt/lists/*
+
+# Create a non-root user
+RUN useradd -m sandboxuser
+USER sandboxuser
+WORKDIR /home/sandboxuser
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+CMD ["python", "--version"]

examples/object_oriented_agentic_approach/resources/diagrams/images/class_diagram_mermaid.png

@@ -0,0 +1,5 @@
+numpy==1.23.5
+pandas==1.5.3
+matplotlib==3.7.2
+seaborn==0.12.2
+scikit-learn==1.2.2

examples/object_oriented_agentic_approach/resources/diagrams/src/uml.mmd

@@ -0,0 +1,37 @@
+# object_oriented_agents/core_classes/agent_signature.py
+from typing import Optional, Dict, Any, List
+from .tool_manager import ToolManager
+
+class AgentSignature:
+    """
+    Encapsulates the logic to produce an agent's 'signature' data:
+    - The developer prompt
+    - The model name
+    - The list of tool definitions
+    """
+
+    def __init__(self, developer_prompt: str, model_name: str, tool_manager: Optional[ToolManager] = None):
+        self.developer_prompt = developer_prompt
+        self.model_name = model_name
+        self.tool_manager = tool_manager
+
+    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)
+        """
+        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 {
+            "developer_prompt": self.developer_prompt,
+            "model_name": self.model_name,
+            "tools": functions
+        }

examples/object_oriented_agentic_approach/resources/docker/dockerfile

@@ -0,0 +1,95 @@
+# object_oriented_agents/core_classes/base_agent.py
+from abc import ABC, abstractmethod
+from typing import Optional
+from .chat_messages import ChatMessages
+from .tool_manager import ToolManager
+from ..utils.logger import get_logger
+from ..services.language_model_interface import LanguageModelInterface
+from .agent_signature import AgentSignature
+
+
+class BaseAgent(ABC):
+    """
+    An abstract base agent that defines the high-level approach to handling user tasks
+    and orchestrating calls to the OpenAI API.
+    """
+
+    def __init__(
+            self,
+            developer_prompt: str,
+            model_name: str,
+            logger=None,
+            language_model_interface: LanguageModelInterface = 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
+
+    @abstractmethod
+    def setup_tools(self) -> None:
+        pass
+
+    def add_context(self, content: str) -> None:
+        self.logger.debug(f"Adding context: {content}")
+        self.messages.add_user_message(content)
+
+    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:
+        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
+        self.add_message(user_task)
+
+        tools = []
+        if tool_call_enabled and self.tool_manager:
+            tools = self.tool_manager.get_tool_definitions()
+            self.logger.debug(f"Tools available: {tools}")
+
+        # Submit to OpenAI
+        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,
+        )
+
+        tool_calls = response.choices[0].message.tool_calls
+        if tool_call_enabled and self.tool_manager and tool_calls:
+            self.logger.debug(f"Tool calls requested: {tool_calls}")
+            return self.tool_manager.handle_tool_call_sequence(
+                response,
+                return_tool_response_as_is,
+                self.messages,
+                self.model_name
+            )
+
+        # No tool call, normal assistant response
+        response_message = response.choices[0].message.content
+        self.messages.add_assistant_message(response_message)
+        self.logger.debug("Task completed successfully.")
+        return response_message
+
+    def signature(self) -> dict:
+        """
+        Return a dictionary with:
+        - The developer prompt
+        - The model name
+        - The tool definitions (function schemas)
+        """
+        signature_obj = AgentSignature(
+            developer_prompt=self.developer_prompt,
+            model_name=self.model_name,
+            tool_manager=self.tool_manager
+        )
+        return signature_obj.to_dict()

examples/object_oriented_agentic_approach/resources/docker/requirements.txt

@@ -0,0 +1,23 @@
+# object_oriented_agents/core_classes/chat_messages.py
+from typing import List, Dict
+
+class ChatMessages:
+    """
+    Stores all messages in a conversation (developer, user, assistant).
+    """
+
+    def __init__(self, developer_prompt: str):
+        self.messages: List[Dict[str, str]] = []
+        self.add_developer_message(developer_prompt)
+
+    def add_developer_message(self, content: str) -> None:
+        self.messages.append({"role": "developer", "content": content})
+
+    def add_user_message(self, content: str) -> None:
+        self.messages.append({"role": "user", "content": content})
+
+    def add_assistant_message(self, content: str) -> None:
+        self.messages.append({"role": "assistant", "content": content})
+
+    def get_messages(self) -> List[Dict[str, str]]:
+        return self.messages