custom

Author: bright-spark

app/backend/approaches-old/__init__.py

@@ -0,0 +1,150 @@
+import json
+import re
+from abc import ABC, abstractmethod
+from typing import Any, AsyncGenerator, Awaitable, Optional, Union, cast
+
+from openai import AsyncStream
+from openai.types.chat import (
+    ChatCompletion,
+    ChatCompletionChunk,
+    ChatCompletionMessageParam,
+)
+
+from approaches.approach import (
+    Approach,
+    ExtraInfo,
+)
+
+
+class ChatApproach(Approach, ABC):
+
+    NO_RESPONSE = "0"
+
+    @abstractmethod
+    async def run_until_final_call(
+        self, messages, overrides, auth_claims, should_stream
+    ) -> tuple[ExtraInfo, Union[Awaitable[ChatCompletion], Awaitable[AsyncStream[ChatCompletionChunk]]]]:
+        pass
+
+    def get_search_query(self, chat_completion: ChatCompletion, user_query: str):
+        response_message = chat_completion.choices[0].message
+
+        if response_message.tool_calls:
+            for tool in response_message.tool_calls:
+                if tool.type != "function":
+                    continue
+                function = tool.function
+                if function.name == "search_sources":
+                    arg = json.loads(function.arguments)
+                    search_query = arg.get("search_query", self.NO_RESPONSE)
+                    if search_query != self.NO_RESPONSE:
+                        return search_query
+        elif query_text := response_message.content:
+            if query_text.strip() != self.NO_RESPONSE:
+                return query_text
+        return user_query
+
+    def extract_followup_questions(self, content: Optional[str]):
+        if content is None:
+            return content, []
+        return content.split("<<")[0], re.findall(r"<<([^>>]+)>>", content)
+
+    async def run_without_streaming(
+        self,
+        messages: list[ChatCompletionMessageParam],
+        overrides: dict[str, Any],
+        auth_claims: dict[str, Any],
+        session_state: Any = None,
+    ) -> dict[str, Any]:
+        extra_info, chat_coroutine = await self.run_until_final_call(
+            messages, overrides, auth_claims, should_stream=False
+        )
+        chat_completion_response: ChatCompletion = await cast(Awaitable[ChatCompletion], chat_coroutine)
+        content = chat_completion_response.choices[0].message.content
+        role = chat_completion_response.choices[0].message.role
+        if overrides.get("suggest_followup_questions"):
+            content, followup_questions = self.extract_followup_questions(content)
+            extra_info.followup_questions = followup_questions
+        # Assume last thought is for generating answer
+        if self.include_token_usage and extra_info.thoughts and chat_completion_response.usage:
+            extra_info.thoughts[-1].update_token_usage(chat_completion_response.usage)
+        chat_app_response = {
+            "message": {"content": content, "role": role},
+            "context": extra_info,
+            "session_state": session_state,
+        }
+        return chat_app_response
+
+    async def run_with_streaming(
+        self,
+        messages: list[ChatCompletionMessageParam],
+        overrides: dict[str, Any],
+        auth_claims: dict[str, Any],
+        session_state: Any = None,
+    ) -> AsyncGenerator[dict, None]:
+        extra_info, chat_coroutine = await self.run_until_final_call(
+            messages, overrides, auth_claims, should_stream=True
+        )
+        chat_coroutine = cast(Awaitable[AsyncStream[ChatCompletionChunk]], chat_coroutine)
+        yield {"delta": {"role": "assistant"}, "context": extra_info, "session_state": session_state}
+
+        followup_questions_started = False
+        followup_content = ""
+        async for event_chunk in await chat_coroutine:
+            # "2023-07-01-preview" API version has a bug where first response has empty choices
+            event = event_chunk.model_dump()  # Convert pydantic model to dict
+            if event["choices"]:
+                # No usage during streaming
+                completion = {
+                    "delta": {
+                        "content": event["choices"][0]["delta"].get("content"),
+                        "role": event["choices"][0]["delta"]["role"],
+                    }
+                }
+                # if event contains << and not >>, it is start of follow-up question, truncate
+                content = completion["delta"].get("content")
+                content = content or ""  # content may either not exist in delta, or explicitly be None
+                if overrides.get("suggest_followup_questions") and "<<" in content:
+                    followup_questions_started = True
+                    earlier_content = content[: content.index("<<")]
+                    if earlier_content:
+                        completion["delta"]["content"] = earlier_content
+                        yield completion
+                    followup_content += content[content.index("<<") :]
+                elif followup_questions_started:
+                    followup_content += content
+                else:
+                    yield completion
+            else:
+                # Final chunk at end of streaming should contain usage
+                # https://cookbook.openai.com/examples/how_to_stream_completions#4-how-to-get-token-usage-data-for-streamed-chat-completion-response
+                if event_chunk.usage and extra_info.thoughts and self.include_token_usage:
+                    extra_info.thoughts[-1].update_token_usage(event_chunk.usage)
+                    yield {"delta": {"role": "assistant"}, "context": extra_info, "session_state": session_state}
+
+        if followup_content:
+            _, followup_questions = self.extract_followup_questions(followup_content)
+            yield {
+                "delta": {"role": "assistant"},
+                "context": {"context": extra_info, "followup_questions": followup_questions},
+            }
+
+    async def run(
+        self,
+        messages: list[ChatCompletionMessageParam],
+        session_state: Any = None,
+        context: dict[str, Any] = {},
+    ) -> dict[str, Any]:
+        overrides = context.get("overrides", {})
+        auth_claims = context.get("auth_claims", {})
+        return await self.run_without_streaming(messages, overrides, auth_claims, session_state)
+
+    async def run_stream(
+        self,
+        messages: list[ChatCompletionMessageParam],
+        session_state: Any = None,
+        context: dict[str, Any] = {},
+    ) -> AsyncGenerator[dict[str, Any], None]:
+        overrides = context.get("overrides", {})
+        auth_claims = context.get("auth_claims", {})
+        return self.run_with_streaming(messages, overrides, auth_claims, session_state)

app/backend/approaches-old/approach.py

@@ -0,0 +1,200 @@
+from typing import Any, Awaitable, List, Optional, Union, cast
+
+from azure.search.documents.aio import SearchClient
+from azure.search.documents.models import VectorQuery
+from openai import AsyncOpenAI, AsyncStream
+from openai.types.chat import (
+    ChatCompletion,
+    ChatCompletionChunk,
+    ChatCompletionMessageParam,
+    ChatCompletionToolParam,
+)
+
+from approaches.approach import DataPoints, ExtraInfo, ThoughtStep
+from approaches.chatapproach import ChatApproach
+from approaches.promptmanager import PromptManager
+from core.authentication import AuthenticationHelper
+
+
+class ChatReadRetrieveReadApproach(ChatApproach):
+    """
+    A multi-step approach that first uses OpenAI to turn the user's question into a search query,
+    then uses Azure AI Search to retrieve relevant documents, and then sends the conversation history,
+    original user question, and search results to OpenAI to generate a response.
+    """
+
+    def __init__(
+        self,
+        *,
+        search_client: SearchClient,
+        auth_helper: AuthenticationHelper,
+        openai_client: AsyncOpenAI,
+        chatgpt_model: str,
+        chatgpt_deployment: Optional[str],  # Not needed for non-Azure OpenAI
+        embedding_deployment: Optional[str],  # Not needed for non-Azure OpenAI or for retrieval_mode="text"
+        embedding_model: str,
+        embedding_dimensions: int,
+        sourcepage_field: str,
+        content_field: str,
+        query_language: str,
+        query_speller: str,
+        prompt_manager: PromptManager,
+        reasoning_effort: Optional[str] = None,
+    ):
+        self.search_client = search_client
+        self.openai_client = openai_client
+        self.auth_helper = auth_helper
+        self.chatgpt_model = chatgpt_model
+        self.chatgpt_deployment = chatgpt_deployment
+        self.embedding_deployment = embedding_deployment
+        self.embedding_model = embedding_model
+        self.embedding_dimensions = embedding_dimensions
+        self.sourcepage_field = sourcepage_field
+        self.content_field = content_field
+        self.query_language = query_language
+        self.query_speller = query_speller
+        self.prompt_manager = prompt_manager
+        self.query_rewrite_prompt = self.prompt_manager.load_prompt("chat_query_rewrite.prompty")
+        self.query_rewrite_tools = self.prompt_manager.load_tools("chat_query_rewrite_tools.json")
+        self.answer_prompt = self.prompt_manager.load_prompt("chat_answer_question.prompty")
+        self.reasoning_effort = reasoning_effort
+        self.include_token_usage = True
+
+    async def run_until_final_call(
+        self,
+        messages: list[ChatCompletionMessageParam],
+        overrides: dict[str, Any],
+        auth_claims: dict[str, Any],
+        should_stream: bool = False,
+    ) -> tuple[ExtraInfo, Union[Awaitable[ChatCompletion], Awaitable[AsyncStream[ChatCompletionChunk]]]]:
+        use_text_search = overrides.get("retrieval_mode") in ["text", "hybrid", None]
+        use_vector_search = overrides.get("retrieval_mode") in ["vectors", "hybrid", None]
+        use_semantic_ranker = True if overrides.get("semantic_ranker") else False
+        use_semantic_captions = True if overrides.get("semantic_captions") else False
+        use_query_rewriting = True if overrides.get("query_rewriting") else False
+        top = overrides.get("top", 3)
+        minimum_search_score = overrides.get("minimum_search_score", 0.0)
+        minimum_reranker_score = overrides.get("minimum_reranker_score", 0.0)
+        filter = self.build_filter(overrides, auth_claims)
+
+        original_user_query = messages[-1]["content"]
+        if not isinstance(original_user_query, str):
+            raise ValueError("The most recent message content must be a string.")
+
+        reasoning_model_support = self.GPT_REASONING_MODELS.get(self.chatgpt_model)
+        if reasoning_model_support and (not reasoning_model_support.streaming and should_stream):
+            raise Exception(
+                f"{self.chatgpt_model} does not support streaming. Please use a different model or disable streaming."
+            )
+
+        query_messages = self.prompt_manager.render_prompt(
+            self.query_rewrite_prompt, {"user_query": original_user_query, "past_messages": messages[:-1]}
+        )
+        tools: List[ChatCompletionToolParam] = self.query_rewrite_tools
+
+        # STEP 1: Generate an optimized keyword search query based on the chat history and the last question
+
+        chat_completion = cast(
+            ChatCompletion,
+            await self.create_chat_completion(
+                self.chatgpt_deployment,
+                self.chatgpt_model,
+                messages=query_messages,
+                overrides=overrides,
+                response_token_limit=self.get_response_token_limit(
+                    self.chatgpt_model, 100
+                ),  # Setting too low risks malformed JSON, setting too high may affect performance
+                temperature=0.0,  # Minimize creativity for search query generation
+                tools=tools,
+                reasoning_effort="low",  # Minimize reasoning for search query generation
+            ),
+        )
+
+        query_text = self.get_search_query(chat_completion, original_user_query)
+
+        # STEP 2: Retrieve relevant documents from the search index with the GPT optimized query
+
+        # If retrieval mode includes vectors, compute an embedding for the query
+        vectors: list[VectorQuery] = []
+        if use_vector_search:
+            vectors.append(await self.compute_text_embedding(query_text))
+
+        results = await self.search(
+            top,
+            query_text,
+            filter,
+            vectors,
+            use_text_search,
+            use_vector_search,
+            use_semantic_ranker,
+            use_semantic_captions,
+            minimum_search_score,
+            minimum_reranker_score,
+            use_query_rewriting,
+        )
+
+        # STEP 3: Generate a contextual and content specific answer using the search results and chat history
+        text_sources = self.get_sources_content(results, use_semantic_captions, use_image_citation=False)
+        messages = self.prompt_manager.render_prompt(
+            self.answer_prompt,
+            self.get_system_prompt_variables(overrides.get("prompt_template"))
+            | {
+                "include_follow_up_questions": bool(overrides.get("suggest_followup_questions")),
+                "past_messages": messages[:-1],
+                "user_query": original_user_query,
+                "text_sources": text_sources,
+            },
+        )
+
+        extra_info = ExtraInfo(
+            DataPoints(text=text_sources),
+            thoughts=[
+                self.format_thought_step_for_chatcompletion(
+                    title="Prompt to generate search query",
+                    messages=query_messages,
+                    overrides=overrides,
+                    model=self.chatgpt_model,
+                    deployment=self.chatgpt_deployment,
+                    usage=chat_completion.usage,
+                    reasoning_effort="low",
+                ),
+                ThoughtStep(
+                    "Search using generated search query",
+                    query_text,
+                    {
+                        "use_semantic_captions": use_semantic_captions,
+                        "use_semantic_ranker": use_semantic_ranker,
+                        "use_query_rewriting": use_query_rewriting,
+                        "top": top,
+                        "filter": filter,
+                        "use_vector_search": use_vector_search,
+                        "use_text_search": use_text_search,
+                    },
+                ),
+                ThoughtStep(
+                    "Search results",
+                    [result.serialize_for_results() for result in results],
+                ),
+                self.format_thought_step_for_chatcompletion(
+                    title="Prompt to generate answer",
+                    messages=messages,
+                    overrides=overrides,
+                    model=self.chatgpt_model,
+                    deployment=self.chatgpt_deployment,
+                    usage=None,
+                ),
+            ],
+        )
+
+        chat_coroutine = cast(
+            Union[Awaitable[ChatCompletion], Awaitable[AsyncStream[ChatCompletionChunk]]],
+            self.create_chat_completion(
+                self.chatgpt_deployment,
+                self.chatgpt_model,
+                messages,
+                overrides,
+                self.get_response_token_limit(self.chatgpt_model, 1024),
+                should_stream,
+            ),
+        )
+        return (extra_info, chat_coroutine)