Added demo for use with Azure

Author: yumano15

examples/japanse-easy-demo-azure/.data/じゅんさい鍋.pdf

@@ -0,0 +1,2 @@
+myenv/
+.data/.indexes

examples/japanse-easy-demo-azure/.data/すり身揚げ.pdf

@@ -0,0 +1,25 @@
+<!-- markdownlint-disable MD013 MD043 -->
+
+# OpenSSA sample
+
+## ライブラリインストール(初回のみ)
+
+```bash
+poetry install
+```
+
+## 実行前手順
+
+- `.env.template`をコピーして`.env`ファイルを作成し、ファイル内の環境変数を記載する
+- `.data` ディレクトリを作成して、そこに回答の情報となるPDFファイルを配置する
+
+## メモ
+
+- 初回実行時に `.data` 内に `.indexes` ディレクトリが作成されインデックスが配置される
+- 初回はその分実行に時間がかかる
+
+## 実行
+
+```bash
+poetry run python main.py
+```

examples/japanse-easy-demo-azure/.data/たこ飯.pdf

@@ -0,0 +1,21 @@
+from llama_index.llms.azure_openai import AzureOpenAI
+from llama_index.embeddings.azure_openai import AzureOpenAIEmbedding as LlamaEmbedding
+from llama_index.core import Settings
+from .config import LMConfig
+
+azure_llama_index_lm = AzureOpenAI(
+    deployment_name=LMConfig.AZURE_OPENAI_GPT_DEPLOYMENT_NAME,
+    api_key=LMConfig.AZURE_OPENAI_API_KEY,
+    azure_endpoint=LMConfig.AZURE_OPENAI_ENDPOINT,
+    api_version=LMConfig.AZURE_OPENAI_API_VERSION,
+)
+
+azure_llama_index_embedding = LlamaEmbedding(
+    deployment_name=LMConfig.AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME,
+    api_key=LMConfig.AZURE_OPENAI_API_KEY,
+    azure_endpoint=LMConfig.AZURE_OPENAI_ENDPOINT,
+    api_version=LMConfig.AZURE_OPENAI_API_VERSION
+)
+
+Settings.llm = azure_llama_index_lm
+Settings.embed_model = azure_llama_index_embedding

examples/japanse-easy-demo-azure/.data/石狩鍋.pdf

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import json
+import time
+from typing import TYPE_CHECKING
+from loguru import logger
+
+from openai import AzureOpenAI
+from openssa.core.util.lm.base import BaseLM
+from .config import LMConfig
+
+
+if TYPE_CHECKING:
+    from openai.types.chat.chat_completion import ChatCompletion
+    from openssa.core.util.lm.base import LMChatHist
+
+
+@dataclass
+class AzureOpenAILM(BaseLM):
+    """OpenAI LM."""
+
+    azure_endpoint: str = ''
+    api_version: str = LMConfig.AZURE_OPENAI_API_VERSION
+    client: AzureOpenAI = field(init=False)
+
+    def __post_init__(self):
+        """Initialize OpenAI client."""
+        self.client: AzureOpenAI = AzureOpenAI(
+            api_key=self.api_key,
+            api_version=self.api_version,
+            azure_endpoint=self.azure_endpoint,
+        )
+
+    @classmethod
+    def from_defaults(cls) -> AzureOpenAILM:
+        """Get OpenAI LM instance with default parameters."""
+        # pylint: disable=unexpected-keyword-arg
+        return cls(
+            model=LMConfig.AZURE_OPENAI_GPT_MODEL_NAME,
+            api_base=None,
+            api_key=LMConfig.AZURE_OPENAI_API_KEY,
+            azure_endpoint=LMConfig.AZURE_OPENAI_ENDPOINT,
+        )
+
+    def call(self, messages: LMChatHist, **kwargs) -> ChatCompletion:
+        """Call OpenAI LM API and return response object."""
+        return self.client.chat.completions.create(
+            messages=messages,
+            model=self.model,
+            seed=kwargs.pop("seed", LMConfig.DEFAULT_SEED),
+            temperature=kwargs.pop("temperature", LMConfig.DEFAULT_TEMPERATURE),
+            **kwargs,
+        )
+
+    def get_response(
+        self,
+        prompt: str,
+        history: LMChatHist | None = None,
+        json_format: bool = False,
+        **kwargs,
+    ) -> str:
+        """Call OpenAI LM API and return response content."""
+        messages: LMChatHist = history or []
+        messages.append({"role": "user", "content": prompt})
+
+        if json_format:
+            kwargs["response_format"] = {"type": "json_object"}
+
+            """
+            The json_format is set to false by default
+            if true, it retries up to five times with amplified retry times.
+            """
+            MAX_RETRIES = 5
+            retries = 0
+            backoff = 1
+
+            while retries < MAX_RETRIES:
+                try:
+                    logger.info(f"RETRIES: {retries}")
+                    return json.loads(
+                        response := self.call(messages, **kwargs)
+                        .choices[0]
+                        .message.content
+                    )
+                except json.decoder.JSONDecodeError:
+                    retries += 1
+                    logger.debug(
+                        f"INVALID JSON, TO BE RETRIED ({retries}/{MAX_RETRIES}):\n{response}"
+                    )
+                    time.sleep(backoff)
+                    backoff *= 2
+
+            raise ValueError("Max retries reached. Unable to parse JSON response.")
+
+        return self.call(messages, **kwargs).choices[0].message.content

examples/japanse-easy-demo-azure/.gitignore

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, TYPE_CHECKING
+
+from openssa.core.knowledge._prompts import knowledge_injection_lm_chat_msgs
+from .azure_openai import AzureOpenAILM
+
+from openssa.core.program_store._prompts import PROGRAM_SEARCH_PROMPT_TEMPLATE
+
+if TYPE_CHECKING:
+    from openssa.core.knowledge.base import Knowledge
+    from openssa.core.programming.base.program import BaseProgram
+    from openssa.core.resource.base import BaseResource
+    from openssa.core.task.task import Task
+    from openssa.core.util.lm.base import BaseLM, LMChatHist
+
+
+@dataclass
+class ProgramStore:
+    """Program Store containing searchable problem-solving Programs."""
+
+    # informative descriptions of stored problem-solving Programs, indexed by name
+    descriptions: dict[str, str] = field(default_factory=dict,
+                                         init=True,
+                                         repr=False,
+                                         hash=None,
+                                         compare=True,
+                                         metadata=None,
+                                         kw_only=False)
+
+    # stored problem-solving Programs, indexed by name
+    programs: dict[str, BaseProgram] = field(default_factory=dict,
+                                             init=True,
+                                             repr=False,
+                                             hash=None,
+                                             compare=True,
+                                             metadata=None,
+                                             kw_only=False)
+
+    # language model for searching among stored problem-solving Programs
+    lm: BaseLM = field(default_factory=AzureOpenAILM.from_defaults,
+                       init=True,
+                       repr=True,
+                       hash=None,
+                       compare=True,
+                       metadata=None,
+                       kw_only=False)
+
+    def add_or_update_program(self, name: str, description: str, program: BaseProgram):
+        """Add or update a Program with its unique identifying name & informative description."""
+        self.descriptions[name]: str = description # type: ignore
+        self.programs[name]: BaseProgram = program # type: ignore
+
+    def find_program(self, task: Task, knowledge: set[Knowledge] | None = None,
+                     adaptations_from_known_programs: dict[str, Any] | None = None) -> BaseProgram | None:
+        """Find a suitable Program for the posed Problem, or return None."""
+        knowledge_lm_hist: LMChatHist | None = (knowledge_injection_lm_chat_msgs(knowledge=knowledge)
+                                                if knowledge
+                                                else None)
+
+        valid_responses: set[str] = set(self.descriptions)
+        valid_responses.add('NONE')
+
+        matching_program_name: str = ''
+        while matching_program_name not in valid_responses:
+            matching_program_name: str = self.lm.get_response(
+                prompt=PROGRAM_SEARCH_PROMPT_TEMPLATE.format(problem=task.ask,
+                                                             resource_overviews={resource.unique_name: resource.overview
+                                                                                 for resource in task.resources},
+                                                             program_descriptions=self.descriptions),
+                history=knowledge_lm_hist)
+
+        if matching_program_name == 'NONE':
+            return None
+
+        adapted_program: BaseProgram = self.programs[matching_program_name].adapt(**(adaptations_from_known_programs or {}))
+        adapted_program.task: Task = task # type: ignore
+        return adapted_program