Scribd
Upload
34 views8 pages

Async Azure OpenAI Integration Example

This document contains code for setting up various clients and services used in an AI assistant application, including OpenAI, Azure Storage, Azure Search, Azure Key Vault, and authentication. It defines routes for the application including asking questions, chatting, and getting configuration details. It also handles errors and formats responses.

Uploaded by

kumarinetra369
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
34 views8 pages

Async Azure OpenAI Integration Example

This document contains code for setting up various clients and services used in an AI assistant application, including OpenAI, Azure Storage, Azure Search, Azure Key Vault, and authentication. It defines routes for the application including asking questions, chatting, and getting configuration details. It also handles errors and formats responses.

Uploaded by

kumarinetra369
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd

import dataclasses

import io
import json
import logging
import mimetypes
import os
from pathlib import Path
from typing import AsyncGenerator, cast

from [Link] import ResourceNotFoundError


from [Link] import DefaultAzureCredential, get_bearer_token_provider
from [Link] import SecretClient
from [Link] import configure_azure_monitor
from [Link] import SearchClient
from [Link] import SearchIndexClient
from [Link] import BlobServiceClient
from openai import APIError, AsyncAzureOpenAI, AsyncOpenAI
from [Link].aiohttp_client import AioHttpClientInstrumentor
from [Link] import OpenTelemetryMiddleware
from [Link] import HTTPXClientInstrumentor
from quart import (
Blueprint,
Quart,
abort,
current_app,
jsonify,
make_response,
request,
send_file,
send_from_directory,
)
from quart_cors import cors

from [Link] import Approach


from [Link] import ChatReadRetrieveReadApproach
from [Link] import
ChatReadRetrieveReadVisionApproach
from [Link] import RetrieveThenReadApproach
from [Link] import RetrieveThenReadVisionApproach
from [Link] import AuthenticationHelper

CONFIG_OPENAI_TOKEN = "openai_token"
CONFIG_CREDENTIAL = "azure_credential"
CONFIG_ASK_APPROACH = "ask_approach"
CONFIG_ASK_VISION_APPROACH = "ask_vision_approach"
CONFIG_CHAT_VISION_APPROACH = "chat_vision_approach"
CONFIG_CHAT_APPROACH = "chat_approach"
CONFIG_BLOB_CONTAINER_CLIENT = "blob_container_client"
CONFIG_AUTH_CLIENT = "auth_client"
CONFIG_GPT4V_DEPLOYED = "gpt4v_deployed"
CONFIG_SEARCH_CLIENT = "search_client"
CONFIG_OPENAI_CLIENT = "openai_client"
ERROR_MESSAGE = """The app encountered an error processing your request.
If you are an administrator of the app, view the full error in the logs. See
[Link]/appservice-logs for more information.
Error type: {error_type}
"""
ERROR_MESSAGE_FILTER = """Your message contains content that was flagged by the
OpenAI content filter."""
bp = Blueprint("routes", __name__, static_folder="static")
# Fix Windows registry issue with mimetypes
mimetypes.add_type("application/javascript", ".js")
mimetypes.add_type("text/css", ".css")

@[Link]("/")
async def index():
return await bp.send_static_file("[Link]")

# Empty page is recommended for login redirect to work.


# See [Link]
lib/msal-browser/docs/[Link]#redirecturi-considerations for more
information
@[Link]("/redirect")
async def redirect():
return ""

@[Link]("/[Link]")
async def favicon():
return await bp.send_static_file("[Link]")

@[Link]("/assets/<path:path>")
async def assets(path):
return await send_from_directory(Path(__file__).resolve().parent / "static" /
"assets", path)

# Serve content files from blob storage from within the app to keep the example
self-contained.
# *** NOTE *** this assumes that the content files are public, or at least that all
users of the app
# can access all the files. This is also slow and memory hungry.
@[Link]("/content/<path>")
async def content_file(path: str):
# Remove page number from path, [Link] -> [Link]
if [Link]("#page=") > 0:
path_parts = [Link]("#page=", 1)
path = path_parts[0]
[Link]("Opening file %s at page %s", path)
blob_container_client = current_app.config[CONFIG_BLOB_CONTAINER_CLIENT]
try:
blob = await blob_container_client.get_blob_client(path).download_blob()
except ResourceNotFoundError:
[Link]("Path not found: %s", path)
abort(404)
if not [Link] or not [Link].has_key("content_settings"):
abort(404)
mime_type = [Link]["content_settings"]["content_type"]
if mime_type == "application/octet-stream":
mime_type = mimetypes.guess_type(path)[0] or "application/octet-stream"
blob_file = [Link]()
await [Link](blob_file)
blob_file.seek(0)
return await send_file(blob_file, mimetype=mime_type, as_attachment=False,
attachment_filename=path)

def error_dict(error: Exception) -> dict:


if isinstance(error, APIError) and [Link] == "content_filter":
return {"error": ERROR_MESSAGE_FILTER}
return {"error": ERROR_MESSAGE.format(error_type=type(error))}

def error_response(error: Exception, route: str, status_code: int = 500):


[Link]("Exception in %s: %s", route, error)
if isinstance(error, APIError) and [Link] == "content_filter":
status_code = 400
return jsonify(error_dict(error)), status_code

@[Link]("/ask", methods=["POST"])
async def ask():
if not request.is_json:
return jsonify({"error": "request must be json"}), 415
request_json = await request.get_json()
context = request_json.get("context", {})
auth_helper = current_app.config[CONFIG_AUTH_CLIENT]
try:
context["auth_claims"] = await
auth_helper.get_auth_claims_if_enabled([Link])
use_gpt4v = [Link]("overrides", {}).get("use_gpt4v", False)
approach: Approach
if use_gpt4v and CONFIG_ASK_VISION_APPROACH in current_app.config:
approach = cast(Approach,
current_app.config[CONFIG_ASK_VISION_APPROACH])
else:
approach = cast(Approach, current_app.config[CONFIG_ASK_APPROACH])
r = await [Link](
request_json["messages"], context=context,
session_state=request_json.get("session_state")
)
return jsonify(r)
except Exception as error:
return error_response(error, "/ask")

class JSONEncoder([Link]):
def default(self, o):
if dataclasses.is_dataclass(o):
return [Link](o)
return super().default(o)

async def format_as_ndjson(r: AsyncGenerator[dict, None]) -> AsyncGenerator[str,


None]:
try:
async for event in r:
yield [Link](event, ensure_ascii=False, cls=JSONEncoder) + "\n"
except Exception as error:
[Link]("Exception while generating response stream: %s", error)
yield [Link](error_dict(error))
@[Link]("/chat", methods=["POST"])
async def chat():
if not request.is_json:
return jsonify({"error": "request must be json"}), 415
request_json = await request.get_json()
context = request_json.get("context", {})
auth_helper = current_app.config[CONFIG_AUTH_CLIENT]
try:
context["auth_claims"] = await
auth_helper.get_auth_claims_if_enabled([Link])
use_gpt4v = [Link]("overrides", {}).get("use_gpt4v", False)
approach: Approach
if use_gpt4v and CONFIG_CHAT_VISION_APPROACH in current_app.config:
approach = cast(Approach,
current_app.config[CONFIG_CHAT_VISION_APPROACH])
else:
approach = cast(Approach, current_app.config[CONFIG_CHAT_APPROACH])

result = await [Link](


request_json["messages"],
stream=request_json.get("stream", False),
context=context,
session_state=request_json.get("session_state"),
)
if isinstance(result, dict):
return jsonify(result)
else:
response = await make_response(format_as_ndjson(result))
[Link] = None # type: ignore
[Link] = "application/json-lines"
return response
except Exception as error:
return error_response(error, "/chat")

# Send [Link] settings to the client UI


@[Link]("/auth_setup", methods=["GET"])
def auth_setup():
auth_helper = current_app.config[CONFIG_AUTH_CLIENT]
return jsonify(auth_helper.get_auth_setup_for_client())

@[Link]("/config", methods=["GET"])
def config():
return jsonify({"showGPT4VOptions": current_app.config[CONFIG_GPT4V_DEPLOYED]})

@bp.before_app_serving
async def setup_clients():
# Replace these with your own values, either in environment variables or
directly here
AZURE_STORAGE_ACCOUNT = [Link]["AZURE_STORAGE_ACCOUNT"]
AZURE_STORAGE_CONTAINER = [Link]["AZURE_STORAGE_CONTAINER"]
AZURE_SEARCH_SERVICE = [Link]["AZURE_SEARCH_SERVICE"]
AZURE_SEARCH_INDEX = [Link]["AZURE_SEARCH_INDEX"]
VISION_SECRET_NAME = [Link]("VISION_SECRET_NAME")
AZURE_KEY_VAULT_NAME = [Link]("AZURE_KEY_VAULT_NAME")
# Shared by all OpenAI deployments
OPENAI_HOST = [Link]("OPENAI_HOST", "azure")
OPENAI_CHATGPT_MODEL = [Link]["AZURE_OPENAI_CHATGPT_MODEL"]
OPENAI_EMB_MODEL = [Link]("AZURE_OPENAI_EMB_MODEL_NAME", "text-embedding-
ada-002")
# Used with Azure OpenAI deployments
AZURE_OPENAI_SERVICE = [Link]("AZURE_OPENAI_SERVICE")
AZURE_OPENAI_GPT4V_DEPLOYMENT = [Link]("AZURE_OPENAI_GPT4V_DEPLOYMENT")
AZURE_OPENAI_GPT4V_MODEL = [Link]("AZURE_OPENAI_GPT4V_MODEL")
AZURE_OPENAI_CHATGPT_DEPLOYMENT = [Link]("AZURE_OPENAI_CHATGPT_DEPLOYMENT")
if OPENAI_HOST == "azure" else None
AZURE_OPENAI_EMB_DEPLOYMENT = [Link]("AZURE_OPENAI_EMB_DEPLOYMENT") if
OPENAI_HOST == "azure" else None
AZURE_VISION_ENDPOINT = [Link]("AZURE_VISION_ENDPOINT", "")
# Used only with non-Azure OpenAI deployments
OPENAI_API_KEY = [Link]("OPENAI_API_KEY")
OPENAI_ORGANIZATION = [Link]("OPENAI_ORGANIZATION")

AZURE_TENANT_ID = [Link]("AZURE_TENANT_ID")
AZURE_USE_AUTHENTICATION = [Link]("AZURE_USE_AUTHENTICATION", "").lower() ==
"true"
AZURE_ENFORCE_ACCESS_CONTROL = [Link]("AZURE_ENFORCE_ACCESS_CONTROL",
"").lower() == "true"
AZURE_SERVER_APP_ID = [Link]("AZURE_SERVER_APP_ID")
AZURE_SERVER_APP_SECRET = [Link]("AZURE_SERVER_APP_SECRET")
AZURE_CLIENT_APP_ID = [Link]("AZURE_CLIENT_APP_ID")
AZURE_AUTH_TENANT_ID = [Link]("AZURE_AUTH_TENANT_ID", AZURE_TENANT_ID)

KB_FIELDS_CONTENT = [Link]("KB_FIELDS_CONTENT", "content")


KB_FIELDS_SOURCEPAGE = [Link]("KB_FIELDS_SOURCEPAGE", "sourcepage")

AZURE_SEARCH_QUERY_LANGUAGE = [Link]("AZURE_SEARCH_QUERY_LANGUAGE", "en-us")


AZURE_SEARCH_QUERY_SPELLER = [Link]("AZURE_SEARCH_QUERY_SPELLER", "lexicon")

USE_GPT4V = [Link]("USE_GPT4V", "").lower() == "true"

# Use the current user identity to authenticate with Azure OpenAI, AI Search
and Blob Storage (no secrets needed,
# just use 'az login' locally, and managed identity when deployed on Azure). If
you need to use keys, use separate AzureKeyCredential instances with the
# keys for each service
# If you encounter a blocking error during a DefaultAzureCredential resolution,
you can exclude the problematic credential by using a parameter (ex.
exclude_shared_token_cache_credential=True)
azure_credential =
DefaultAzureCredential(exclude_shared_token_cache_credential=True)

# Set up clients for AI Search and Storage


search_client = SearchClient(
endpoint=f"[Link]
index_name=AZURE_SEARCH_INDEX,
credential=azure_credential,
)
search_index_client = SearchIndexClient(
endpoint=f"[Link]
credential=azure_credential,
)
blob_client = BlobServiceClient(
account_url=f"[Link]
credential=azure_credential
)
blob_container_client =
blob_client.get_container_client(AZURE_STORAGE_CONTAINER)

# Set up authentication helper


auth_helper = AuthenticationHelper(
search_index=(await search_index_client.get_index(AZURE_SEARCH_INDEX)) if
AZURE_USE_AUTHENTICATION else None,
use_authentication=AZURE_USE_AUTHENTICATION,
server_app_id=AZURE_SERVER_APP_ID,
server_app_secret=AZURE_SERVER_APP_SECRET,
client_app_id=AZURE_CLIENT_APP_ID,
tenant_id=AZURE_AUTH_TENANT_ID,
require_access_control=AZURE_ENFORCE_ACCESS_CONTROL,
)

vision_key = None
if VISION_SECRET_NAME and AZURE_KEY_VAULT_NAME: # Cognitive vision keys are
stored in keyvault
key_vault_client = SecretClient(
vault_url=f"[Link]
credential=azure_credential
)
vision_secret = await key_vault_client.get_secret(VISION_SECRET_NAME)
vision_key = vision_secret.value
await key_vault_client.close()

# Used by the OpenAI SDK


openai_client: AsyncOpenAI

if OPENAI_HOST == "azure":
token_provider = get_bearer_token_provider(azure_credential,
"[Link]
# Store on [Link] for later use inside requests
openai_client = AsyncAzureOpenAI(
api_version="2023-07-01-preview",
azure_endpoint=f"[Link]
azure_ad_token_provider=token_provider,
)
else:
openai_client = AsyncOpenAI(
api_key=OPENAI_API_KEY,
organization=OPENAI_ORGANIZATION,
)

current_app.config[CONFIG_OPENAI_CLIENT] = openai_client
current_app.config[CONFIG_SEARCH_CLIENT] = search_client
current_app.config[CONFIG_BLOB_CONTAINER_CLIENT] = blob_container_client
current_app.config[CONFIG_AUTH_CLIENT] = auth_helper

current_app.config[CONFIG_GPT4V_DEPLOYED] = bool(USE_GPT4V)

# Various approaches to integrate GPT and external knowledge, most applications


will use a single one of these patterns
# or some derivative, here we include several for exploration purposes
current_app.config[CONFIG_ASK_APPROACH] = RetrieveThenReadApproach(
search_client=search_client,
openai_client=openai_client,
auth_helper=auth_helper,
chatgpt_model=OPENAI_CHATGPT_MODEL,
chatgpt_deployment=AZURE_OPENAI_CHATGPT_DEPLOYMENT,
embedding_model=OPENAI_EMB_MODEL,
embedding_deployment=AZURE_OPENAI_EMB_DEPLOYMENT,
sourcepage_field=KB_FIELDS_SOURCEPAGE,
content_field=KB_FIELDS_CONTENT,
query_language=AZURE_SEARCH_QUERY_LANGUAGE,
query_speller=AZURE_SEARCH_QUERY_SPELLER,
)

if USE_GPT4V:
if vision_key is None:
raise ValueError("Vision key must be set (in Key Vault) to use the
vision approach.")

current_app.config[CONFIG_ASK_VISION_APPROACH] =
RetrieveThenReadVisionApproach(
search_client=search_client,
openai_client=openai_client,
blob_container_client=blob_container_client,
auth_helper=auth_helper,
vision_endpoint=AZURE_VISION_ENDPOINT,
vision_key=vision_key,
gpt4v_deployment=AZURE_OPENAI_GPT4V_DEPLOYMENT,
gpt4v_model=AZURE_OPENAI_GPT4V_MODEL,
embedding_model=OPENAI_EMB_MODEL,
embedding_deployment=AZURE_OPENAI_EMB_DEPLOYMENT,
sourcepage_field=KB_FIELDS_SOURCEPAGE,
content_field=KB_FIELDS_CONTENT,
query_language=AZURE_SEARCH_QUERY_LANGUAGE,
query_speller=AZURE_SEARCH_QUERY_SPELLER,
)

current_app.config[CONFIG_CHAT_VISION_APPROACH] =
ChatReadRetrieveReadVisionApproach(
search_client=search_client,
openai_client=openai_client,
blob_container_client=blob_container_client,
auth_helper=auth_helper,
vision_endpoint=AZURE_VISION_ENDPOINT,
vision_key=vision_key,
gpt4v_deployment=AZURE_OPENAI_GPT4V_DEPLOYMENT,
gpt4v_model=AZURE_OPENAI_GPT4V_MODEL,
embedding_model=OPENAI_EMB_MODEL,
embedding_deployment=AZURE_OPENAI_EMB_DEPLOYMENT,
sourcepage_field=KB_FIELDS_SOURCEPAGE,
content_field=KB_FIELDS_CONTENT,
query_language=AZURE_SEARCH_QUERY_LANGUAGE,
query_speller=AZURE_SEARCH_QUERY_SPELLER,
)

current_app.config[CONFIG_CHAT_APPROACH] = ChatReadRetrieveReadApproach(
search_client=search_client,
openai_client=openai_client,
auth_helper=auth_helper,
chatgpt_model=OPENAI_CHATGPT_MODEL,
chatgpt_deployment=AZURE_OPENAI_CHATGPT_DEPLOYMENT,
embedding_model=OPENAI_EMB_MODEL,
embedding_deployment=AZURE_OPENAI_EMB_DEPLOYMENT,
sourcepage_field=KB_FIELDS_SOURCEPAGE,
content_field=KB_FIELDS_CONTENT,
query_language=AZURE_SEARCH_QUERY_LANGUAGE,
query_speller=AZURE_SEARCH_QUERY_SPELLER,
)

@bp.after_app_serving
async def close_clients():
await current_app.config[CONFIG_SEARCH_CLIENT].close()
await current_app.config[CONFIG_BLOB_CONTAINER_CLIENT].close()

def create_app():
app = Quart(__name__)
app.register_blueprint(bp)

if [Link]("APPLICATIONINSIGHTS_CONNECTION_STRING"):
configure_azure_monitor()
# This tracks HTTP requests made by aiohttp:
AioHttpClientInstrumentor().instrument()
# This tracks HTTP requests made by httpx/openai:
HTTPXClientInstrumentor().instrument()
# This middleware tracks app route requests:
app.asgi_app = OpenTelemetryMiddleware(app.asgi_app) # type:
ignore[method-assign]

# Level should be one of


[Link]
default_level = "INFO" # In development, log more verbosely
if [Link]("WEBSITE_HOSTNAME"): # In production, don't log as heavily
default_level = "WARNING"
[Link](level=[Link]("APP_LOG_LEVEL", default_level))

if allowed_origin := [Link]("ALLOWED_ORIGIN"):
[Link]("CORS enabled for %s", allowed_origin)
cors(app, allow_origin=allowed_origin, allow_methods=["GET", "POST"])
return app

You might also like