Updates util.py

examples/customer_service_streaming/src/swarm/swarm.py

@@ -1,33 +1,35 @@
 import json
 from openai import OpenAI
-from src.tasks.task import Task,EvaluationTask
+from src.tasks.task import Task, EvaluationTask
 from src.swarm.engines.assistants_engine import AssistantsEngine
 from src.swarm.engines.local_engine import LocalEngine
 from configs.general import Colors, tasks_path
 
+# This class represents the main control unit for deploying and managing tasks within the swarm system.
+
+
 class Swarm:
-    def __init__(self,engine_name,tasks=[], persist=False):
+    def __init__(self, engine_name, tasks=[], persist=False):
         self.tasks = tasks
         self.engine_name = engine_name
         self.engine = None
         self.persist = persist
 
-
-    def deploy(self, test_mode=False,test_file_paths=None):
+    def deploy(self, test_mode=False, test_file_paths=None):
         """
         Processes all tasks in the order they are listed in self.tasks.
         """
         client = OpenAI()
-        #Initialize swarm first
+        # Initialize swarm first
         if self.engine_name == 'assistants':
             print(f"{Colors.GREY}Selected engine: Assistants{Colors.ENDC}")
-            self.engine = AssistantsEngine(client,self.tasks)
-            self.engine.deploy(client,test_mode,test_file_paths)
+            self.engine = AssistantsEngine(client, self.tasks)
+            self.engine.deploy(client, test_mode, test_file_paths)
 
-        elif self.engine_name =='local':
+        elif self.engine_name == 'local':
             print(f"{Colors.GREY}Selected engine: Local{Colors.ENDC}")
-            self.engine = LocalEngine(client,self.tasks, persist=self.persist)
-            self.engine.deploy(client,test_mode,test_file_paths)
+            self.engine = LocalEngine(client, self.tasks, persist=self.persist)
+            self.engine.deploy(client, test_mode, test_file_paths)
 
     def load_tasks(self):
         self.tasks = []
@@ -42,4 +44,3 @@ def load_tasks(self):
 
     def add_task(self, task):
         self.tasks.append(task)
-

examples/customer_service_streaming/src/swarm/tool.py

@@ -1,21 +1,25 @@
 from pydantic import BaseModel, Field
 from typing import Dict, List, Optional, Literal
 
+
 class Parameter(BaseModel):
     type: str
     description: Optional[str] = None
     enum: Optional[List[str]] = Field(None, alias='choices')
 
+
 class FunctionParameters(BaseModel):
     type: Literal['object']  # Ensuring it's always 'object'
     properties: Dict[str, Parameter] = {}
     required: Optional[List[str]] = None
 
+
 class FunctionTool(BaseModel):
     name: str
     description: Optional[str]
     parameters: FunctionParameters
 
+
 class Tool(BaseModel):
     type: str
     function: Optional[FunctionTool]

swarm/core.py

@@ -94,7 +94,8 @@ def handle_tool_calls(
         debug: bool,
     ) -> Response:
         function_map = {f.__name__: f for f in functions}
-        partial_response = Response(messages=[], agent=None, context_variables={})
+        partial_response = Response(
+            messages=[], agent=None, context_variables={})
 
         for tool_call in tool_calls:
             name = tool_call.function.name
@@ -111,7 +112,8 @@ def handle_tool_calls(
                 )
                 continue
             args = json.loads(tool_call.function.arguments)
-            debug_print(debug, f"Processing tool call: {name} with arguments {args}")
+            debug_print(
+                debug, f"Processing tool call: {name} with arguments {args}")
 
             func = function_map[name]
             # pass context_variables to agent functions
@@ -186,7 +188,8 @@ def run_and_stream(
                 merge_chunk(message, delta)
             yield {"delim": "end"}
 
-            message["tool_calls"] = list(message.get("tool_calls", {}).values())
+            message["tool_calls"] = list(
+                message.get("tool_calls", {}).values())
             if not message["tool_calls"]:
                 message["tool_calls"] = None
             debug_print(debug, "Received completion:", message)

swarm/util.py

@@ -29,6 +29,17 @@ def merge_chunk(final_response: dict, delta: dict) -> None:
 
 
 def function_to_json(func) -> dict:
+    """
+    Converts a Python function into a JSON-serializable dictionary
+    that describes the function's signature, including its name,
+    description, and parameters.
+
+    Args:
+        func: The function to be converted.
+
+    Returns:
+        A dictionary representing the function's signature in JSON format.
+    """
     type_map = {
         str: "string",
         int: "integer",