Python sample for assistant skills

samples/assistant/README.md

@@ -55,6 +55,26 @@ app.generic('AddTodo', {
 })
 ```
 
+Python example:
+
+```py
+
+skills = func.Blueprint()
+todo_manager = CreateTodoManager()
+
+@skills.function_name("AddTodo")
+@skills.generic_trigger(arg_name="taskDescription", type="assistantSkillTrigger", data_type=func.DataType.STRING, functionDescription="Create a new todo task")
+def add_todo(taskDescription: str) -> None:
+    if not taskDescription:
+        raise ValueError("Task description cannot be empty")
+
+    logging.info(f"Adding todo: {taskDescription}")
+
+    todo_id = str(uuid.uuid4())[0:6]
+    todo_manager.add_todo(TodoItem(id=todo_id, task=taskDescription))
+    return
+```
+
 The `AssistantSkillTrigger` attribute requires a `FunctionDescription` string value, which is text describing what the function does.
 This is critical for the AI assistant to be able to invoke the skill at the right time.
 The name of the function parameter (e.g., `taskDescription`) is also an important hint to the AI assistant about what kind of information to provide to the skill.
@@ -70,6 +90,7 @@ The sample is available in the following language stacks:
 
 * [C# on the out-of-process worker](csharp-ooproc)
 * [nodejs](nodejs)
+* [python](python) - supported on host runtime version >= 4.34.0.0
 
 Please refer to the [root README](../../README.md#requirements) for common prerequisites that apply to all samples.
 

samples/assistant/python/.funcignore

@@ -0,0 +1,8 @@
+.git*
+.vscode
+__azurite_db*__.json
+__blobstorage__
+__queuestorage__
+local.settings.json
+test
+.venv

samples/assistant/python/assistant_apis.py

@@ -0,0 +1,41 @@
+import json
+import azure.functions as func
+
+apis = func.Blueprint()
+
+
+@apis.function_name("CreateAssistant")
+@apis.route(route="assistants/{assistantId}", methods=["PUT"])
+@apis.generic_output_binding(arg_name="requests", type="assistantCreate", data_type=func.DataType.STRING)
+def create_assistant(req: func.HttpRequest, requests: func.Out[str]) -> func.HttpResponse:
+    assistantId = req.route_params.get("assistantId")
+    instructions = """
+            Don't make assumptions about what values to plug into functions.
+            Ask for clarification if a user request is ambiguous.
+            """
+    create_request = {
+        "id": assistantId,
+        "instructions": instructions
+    }
+    requests.set(json.dumps(create_request))
+    response_json = {"assistantId": assistantId}
+    return func.HttpResponse(json.dumps(response_json), status_code=202, mimetype="application/json")
+
+
+@apis.function_name("PostUserQuery")
+@apis.route(route="assistants/{assistantId}", methods=["POST"])
+@apis.generic_output_binding(arg_name="requests", type="assistantPost", data_type=func.DataType.STRING, id="{assistantId}", model="%CHAT_MODEL_DEPLOYMENT_NAME%")
+def post_user_query(req: func.HttpRequest, requests: func.Out[str]) -> func.HttpResponse:
+    userMessage = req.get_body().decode("utf-8")
+    if not userMessage:
+        return func.HttpResponse(json.dumps({"message": "Request body is empty"}), status_code=400, mimetype="application/json")
+
+    requests.set(json.dumps({"userMessage": userMessage}))
+    return func.HttpResponse(status_code=202)
+
+
+@apis.function_name("GetChatState")
+@apis.route(route="assistants/{assistantId}", methods=["GET"])
+@apis.generic_input_binding(arg_name="state", type="assistantQuery", data_type=func.DataType.STRING, id="{assistantId}", timestampUtc="{Query.timestampUTC}")
+def get_chat_state(req: func.HttpRequest, state: str) -> func.HttpResponse:
+    return func.HttpResponse(state, status_code=200, mimetype="application/json")

samples/assistant/python/assistant_skills.py

@@ -0,0 +1,31 @@
+import json
+import logging
+import uuid
+import azure.functions as func
+
+from todo_manager import CreateTodoManager, TodoItem
+
+skills = func.Blueprint()
+
+todo_manager = CreateTodoManager()
+
+
+@skills.function_name("AddTodo")
+@skills.generic_trigger(arg_name="taskDescription", type="assistantSkillTrigger", data_type=func.DataType.STRING, functionDescription="Create a new todo task")
+def add_todo(taskDescription: str) -> None:
+    if not taskDescription:
+        raise ValueError("Task description cannot be empty")
+
+    logging.info(f"Adding todo: {taskDescription}")
+
+    todo_id = str(uuid.uuid4())[0:6]
+    todo_manager.add_todo(TodoItem(id=todo_id, task=taskDescription))
+    return
+
+
+@skills.function_name("GetTodos")
+@skills.generic_trigger(arg_name="inputIgnored", type="assistantSkillTrigger", data_type=func.DataType.STRING, functionDescription="Fetch the list of previously created todo tasks")
+def get_todos(inputIgnored: str) -> str:
+    logging.info("Fetching list of todos")
+    results = todo_manager.get_todos()
+    return json.dumps(results)

samples/assistant/python/function_app.py

@@ -0,0 +1,9 @@
+import azure.functions as func
+
+from assistant_apis import apis
+from assistant_skills import skills
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+app.register_functions(apis)
+app.register_functions(skills)

samples/assistant/python/host.json

@@ -0,0 +1,18 @@
+{
+  "version": "2.0",
+  "logging": {
+    "logLevel": {
+      "Microsoft.Azure.WebJobs.Extensions.OpenAI": "Information"
+    }
+  },
+  "extensionBundle": {
+    "id": "Microsoft.Azure.Functions.ExtensionBundle.Preview",
+    "version": "[4.*, 5.0.0)"
+  },
+  "extensions": {
+    "openai": {
+      "storageConnectionName": "AzureWebJobsStorage",
+      "collectionName": "SampleChatState"
+    }
+  }
+}

samples/assistant/python/local.settings.json

@@ -0,0 +1,13 @@
+{
+  "IsEncrypted": false,
+  "Values": {
+    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
+    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
+    "FUNCTIONS_WORKER_RUNTIME": "python",
+    "CosmosDbConnectionString": "Local Cosmos DB emulator or Cosmos DB in Azure connection string",
+    "CosmosDatabaseName": "testdb",
+    "CosmosContainerName": "my-todos",
+    "AZURE_OPENAI_ENDPOINT": "https://<resource-name>.openai.azure.com/",
+    "CHAT_MODEL_DEPLOYMENT_NAME": "gpt-3.5-turbo"
+  }
+}

samples/assistant/python/requirements.txt

@@ -0,0 +1,6 @@
+# DO NOT include azure-functions-worker in this file
+# The Python Worker is managed by Azure Functions platform
+# Manually managing azure-functions-worker may cause unexpected issues
+
+azure-functions
+azure-cosmos

samples/assistant/python/todo_manager.py

@@ -0,0 +1,68 @@
+import abc
+import logging
+import os
+
+from azure.cosmos import CosmosClient
+from azure.cosmos import PartitionKey
+
+class TodoItem:
+    def __init__(self, id, task):
+        self.id = id
+        self.task = task
+
+
+class ITodoManager(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def add_todo(self, todo: TodoItem):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def get_todos(self):
+        raise NotImplementedError()
+
+
+class InMemoryTodoManager(ITodoManager):
+    def __init__(self):
+        self.todos = []
+
+    def add_todo(self, todo: TodoItem):
+        self.todos.append(todo)
+
+    def get_todos(self):
+        return [item.__dict__ for item in self.todos]
+
+
+class CosmosDbTodoManager(ITodoManager):
+    def __init__(self, cosmos_client: CosmosClient):
+        self.cosmos_client = cosmos_client
+        cosmos_database_name = os.environ.get("CosmosDatabaseName")
+        cosmos_container_name = os.environ.get("CosmosContainerName")
+
+        if not cosmos_database_name or not cosmos_container_name:
+            raise ValueError("CosmosDatabaseName and CosmosContainerName must be set as environment variables or in local.settings.json")
+
+        self.database = self.cosmos_client.create_database_if_not_exists(cosmos_database_name)
+        self.container = self.database.create_container_if_not_exists(id=cosmos_container_name, partition_key=PartitionKey(path="/id"))
+
+    def add_todo(self, todo: TodoItem):
+        logging.info(
+            f"Adding todo ID = {todo.id} to container '{self.container.id}'.")
+        self.container.create_item(todo.__dict__)
+
+    def get_todos(self):
+        logging.info(
+            f"Getting all todos from container '{self.container.id}'.")
+        results = [item for item in self.container.query_items(
+            "SELECT * FROM c", enable_cross_partition_query=True)]
+        logging.info(
+            f"Found {len(results)} todos in container '{self.container.id}'.")
+        return results
+
+
+def CreateTodoManager() -> ITodoManager:
+    if not os.environ.get("CosmosDbConnectionString"):
+        return InMemoryTodoManager()
+    else:
+        cosmos_client = CosmosClient.from_connection_string(
+            os.environ["CosmosDbConnectionString"])
+        return CosmosDbTodoManager(cosmos_client)