Add script to remove autogen (#33)

pyproject.toml

@@ -21,6 +21,7 @@ azure-core = "^1.30.2"
 drawsvg = {extras = ["all"], version = "^2.4.0"}
 svgpathtools = "^1.6.1"
 azure-cosmos = "^4.7.0"
+instructor = "^1.4.2"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.0"

src/autogen_planner.py

@@ -1,17 +1,18 @@
 import json
 import re
-from typing import List, Callable, Optional, Union
-from datetime import datetime
 from dataclasses import dataclass
-from dataclasses_json import dataclass_json
-from botbuilder.schema import Attachment
+from datetime import datetime
+from typing import Callable, List, Optional, Union
+
+from autogen import Agent, ChatResult, GroupChat, GroupChatManager, runtime_logging
 from botbuilder.core import CardFactory, TurnContext
-from teams.ai.prompts import Message
-from teams.ai.planners import Planner, Plan, PredictedSayCommand
-from autogen import Agent, GroupChat, GroupChatManager, ChatResult, runtime_logging
+from botbuilder.schema import Attachment
+from config import Config
+from dataclasses_json import dataclass_json
 from state import AppTurnState
+from teams.ai.planners import Plan, Planner, PredictedSayCommand
+from teams.ai.prompts import Message
 from teams_user_proxy import TeamsUserProxy
-from config import Config
 
 
 @dataclass_json
@@ -37,10 +38,13 @@ def __init__(
         self.message_builder = message_builder
         super().__init__()
 
-    async def begin_task(self, context, state: AppTurnState):
+    async def begin_task(self, context: TurnContext, state: AppTurnState):
         return await self.continue_task(context, state)
 
-    async def continue_task(self, context, state: AppTurnState):
+    async def continue_task(self, context: TurnContext, state: AppTurnState):
+        return await self.run_with_autogen(context, state)
+
+    async def run_with_autogen(self, context: TurnContext, state: AppTurnState):
         user_proxy = TeamsUserProxy(
             name="User",
             system_message="A human admin. This agent is a proxy for the user. This agent can help answer questions too.",
@@ -95,6 +99,7 @@ async def continue_task(self, context, state: AppTurnState):
         )
 
         if Config.ENABLE_RUNTIME_LOGGING:
+            print('Logging is enabled')
             runtime_logging.start()
 
         chat_result = await user_proxy.a_initiate_chat(
@@ -157,7 +162,6 @@ async def continue_task(self, context, state: AppTurnState):
             ]
         )
 
-
 def create_chat_history_ac(message: ChatResult) -> Attachment:
     facts = []
     for value in message.chat_history:

src/bot.py

@@ -5,32 +5,33 @@
 Description: initialize the app and listen for `message` activitys
 """
 
-from typing import Dict
 import sys
 import traceback
-from autogen import GroupChat, Agent
-from botbuilder.schema import Activity, ActivityTypes
+from typing import Dict
 
-from botbuilder.core import TurnContext, MemoryStorage, InvokeResponse
+from autogen import Agent, GroupChat
+from autogen_planner import AutoGenPlanner, PredictedSayCommandWithAttachments
+from autogen_utils import (
+    AppTurnStateConversationState,
+    TurnChatContext,
+    TypingCapability,
+)
+from botbuilder.core import InvokeResponse, MemoryStorage, TurnContext
+from botbuilder.schema import Activity, ActivityTypes
+from config import Config
+from cosmos_memory_storage import CosmosDbPartitionedStorage
+from privacy_review_assistant_group import PrivacyReviewAssistantGroup
+from state import AppTurnState
 from teams import Application, ApplicationOptions, TeamsAdapter
 from teams.ai import AIOptions
-from teams.ai.actions import ActionTypes, ActionTurnContext
+from teams.ai.actions import ActionTurnContext, ActionTypes
 from teams.feedback_loop_data import FeedbackLoopData
 from teams.teams_attachment_downloader.teams_attachment_downloader import (
     TeamsAttachmentDownloader,
 )
 from teams.teams_attachment_downloader.teams_attachment_downloader_options import (
     TeamsAttachmentDownloaderOptions,
 )
-from autogen_planner import AutoGenPlanner, PredictedSayCommandWithAttachments
-from autogen_utils import AppTurnStateConversationState, TypingCapability, TurnChatContext
-
-# from botbuilder.azure import BlobStorage, BlobStorageSettings
-from privacy_review_assistant_group import PrivacyReviewAssistantGroup
-
-from config import Config
-from cosmos_memory_storage import CosmosDbPartitionedStorage
-from state import AppTurnState
 from threat_model_file_utils import (
     get_threat_model_image_file,
     get_threat_model_xml_file,
@@ -39,11 +40,6 @@
 config = Config()
 llm_config = config.build_llm_config()
 
-if config.OPENAI_KEY is None and config.AZURE_OPENAI_KEY is None:
-    raise RuntimeError(
-        "Unable to build LLM config - please check that OPENAI_KEY or AZURE_OPENAI_KEY is set."
-    )
-
 cosmos_config = config.build_cosmos_db_config()
 storage = (
     MemoryStorage()
@@ -155,6 +151,13 @@ async def set_to_xml_multi_prompt(context: TurnContext, state: AppTurnState):
     await context.send_activity("Ready to use multi prompt XML evaluator")
     return True
 
+@app.message("/useNoAutogen")
+async def set_to_no_autogen(context: TurnContext, state: AppTurnState):
+    state.conversation.threat_model_evaluator = "no_autogen"
+    await state.save(context)
+    await context.send_activity("Ready to skip autogen evaluator")
+    return True
+
 
 @app.activity("invoke")
 async def feedback_loop(context: TurnContext, state: AppTurnState):

src/models.py

@@ -0,0 +1,130 @@
+import json
+from typing import Annotated, Dict, List, Literal, Union
+
+from pydantic import BaseModel
+from svg_to_png.lib.ThreatModel import User_Friendly_Block_Types
+
+Hints_To_Send = Union[List[User_Friendly_Block_Types], Literal["all"]]
+
+class Spec(BaseModel):
+    id: int
+    spec: str
+    instructions_to_solve: str
+    improvement_hints: str
+    hints_to_send: Hints_To_Send = "all"
+
+    def build_instruction(self):
+        return f"""Spec id {self.id} - {self.spec}\n{self.instructions_to_solve}
+If the criteria is not met, then {self.improvement_hints}
+"""
+
+
+class SpecAnswer(BaseModel):
+    spec_id: Annotated[int, "The spec id to answer"]
+    detailed_answer: Annotated[
+        str,
+        "Does the threat model meet the spec criteria? Why or why not? Be helpful and specific.",
+    ]
+    steps_to_improve: Annotated[
+        str,
+        "What are exact steps to improve the threat model to meet the spec criteria? Use 'None' if no steps to improve",
+    ]
+    tag: Annotated[
+        Union[
+            Annotated[Literal["green"], "Spec criteria is met"],
+            Annotated[Literal["red"], "Items needs to be fixed to meet criteria"],
+            Annotated[Literal["yellow"], "Criteria is met but can be improved"],
+        ],
+        "The tag of the spec answer",
+    ]
+
+
+class SpecResult(BaseModel):
+    spec: Spec
+    answer: SpecAnswer
+
+
+
+def load_specs_from_json(file_path: str) -> List[Spec]:
+    with open(file_path, "r") as file:
+        specs_data = json.load(file)
+    return [Spec(**spec) for spec in specs_data]
+
+tag_with_headers = {"green": "✅", "red": "❌", "yellow": "⚠️"}
+
+def build_container_for_answer(spec: Spec, spec_answer: SpecAnswer):
+    answer_items = [
+        {
+            "type": "TextBlock",
+            "text": spec_answer.detailed_answer,
+            "wrap": True,
+        }
+    ]
+    if spec_answer.steps_to_improve != "None" and spec_answer.steps_to_improve:
+        answer_items.append(
+            {
+                "type": "TextBlock",
+                "text": "Steps to improve:",
+                "wrap": True,
+                "weight": "Bolder",
+            }
+        )
+        answer_items.append(
+            {
+                "type": "TextBlock",
+                "text": spec_answer.steps_to_improve,
+                "wrap": True,
+            }
+        )
+    return {
+        "type": "Container",
+        "items": [
+            {
+                "type": "TextBlock",
+                "text": spec.spec,
+                "wrap": True,
+                "weight": "Bolder",
+            },
+            {
+                "type": "ColumnSet",
+                "columns": [
+                    {
+                        "type": "Column",
+                        "width": "auto",
+                        "items": [
+                            {
+                                "type": "TextBlock",
+                                "text": tag_with_headers[spec_answer.tag],
+                                "wrap": True,
+                            }
+                        ],
+                    },
+                    {
+                        "type": "Column",
+                        "width": "stretch",
+                        "items": answer_items,
+                    },
+                ],
+            },
+        ],
+        "separator": True,
+    }
+
+
+def build_adaptive_card(spec_answer_containers: List[Dict]):
+    body = [
+        {
+            "type": "TextBlock",
+            "text": "Threat model review",
+            "wrap": True,
+            "weight": "Bolder",
+            "style": "heading",
+        },
+        *spec_answer_containers,
+    ]
+    return {
+        "type": "AdaptiveCard",
+        "$schema": "https://adaptivecards.io/schemas/adaptive-card.json",
+        "version": "1.5",
+        "body": body,
+    }