我有用 Visual Studio 代码开发的 python 应用程序 Flask 机器人,具有各自的 requests.txt 和 template/index.html,经过调试并在本地工作正常,创建了 Linux azure Web 应用程序,当我尝试通过 viz studio 代码或 GitHub 部署进行部署时已成功,但无法运行机器人,收到应用程序错误。尝试了多次都没有成功…
e # application.py
import os
import sys
import logging
from typing import Tuple, Optional, List, Dict
from flask import Flask, render_template, request, jsonify
from azure.core.credentials import AzureKeyCredential
from azure.search.documents import SearchClient
import openai
from datetime import datetime
# Initialize Flask application
application = Flask(__name__)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler('app.log')
]
)
logger = logging.getLogger(__name__)
class Config:
"""Configuration management class"""
# Azure Search Configuration
SEARCH_SERVICE = os.environ.get('SEARCH_SERVICE', 'searchmitkoman')
SEARCH_KEY = os.environ.get('SEARCH_KEY')
SEARCH_INDEX = os.environ.get('SEARCH_INDEX', 'vector-1730871636072')
CONTENT_FIELD = 'chunk'
SOURCE_FIELD = 'title'
# Azure OpenAI Configuration
AZURE_OPENAI_ENDPOINT = os.environ.get('AZURE_OPENAI_ENDPOINT')
AZURE_OPENAI_KEY = os.environ.get('AZURE_OPENAI_KEY')
AZURE_OPENAI_DEPLOYMENT = os.environ.get('AZURE_OPENAI_DEPLOYMENT', 'deployment_four')
@classmethod
def validate(cls) -> bool:
"""Validate required environment variables"""
required_vars = ['SEARCH_KEY', 'AZURE_OPENAI_ENDPOINT', 'AZURE_OPENAI_KEY']
missing_vars = [var for var in required_vars if not getattr(cls, var)]
if missing_vars:
logger.error(f"Missing required environment variables: {', '.join(missing_vars)}")
return False
return True
# Initialize OpenAI client
openai.api_type = "azure"
openai.api_base = Config.AZURE_OPENAI_ENDPOINT
openai.api_version = "2023-05-15"
openai.api_key = Config.AZURE_OPENAI_KEY
def perform_search(query: str) -> Tuple[Optional[str], Optional[List[str]]]:
"""
Execute search using Azure Cognitive Search
Args:
query (str): Search query
Returns:
Tuple[Optional[str], Optional[List[str]]]: Content and references
"""
try:
endpoint = f"https://{Config.SEARCH_SERVICE}.search.windows.net"
credential = AzureKeyCredential(Config.SEARCH_KEY)
client = SearchClient(
endpoint=endpoint,
index_name=Config.SEARCH_INDEX,
credential=credential
)
results = list(client.search(
search_text=query,
select=f"{Config.SOURCE_FIELD},{Config.CONTENT_FIELD}",
top=3
))
if not results:
logger.warning(f"No results found for query: {query}")
return None, None
content = "\n".join([f"{doc[Config.SOURCE_FIELD]}: {doc[Config.CONTENT_FIELD]}"
for doc in results])
references = list(set([doc[Config.SOURCE_FIELD] for doc in results]))
return content, references
except Exception as e:
logger.error(f"Search error: {str(e)}")
return None, None
def generate_answer(conversation: List[Dict[str, str]]) -> str:
"""
@application.route('/health')
def health_check() -> tuple:
"""Health check endpoint"""
status = {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"config_valid": Config.validate()
}
return jsonify(status), 200
@application.route('/', methods=['GET', 'POST'])
def index() -> str:
"""Main application route"""
try:
if request.method == 'POST':
prompt = request.form.get('prompt', '').strip()
if not prompt:
return render_template('index.html', error="Please enter a question.")
# Validate configuration
if not Config.validate():
return render_template('index.html',
error="Application configuration error. Please contact support.")
# Perform search
content, references = perform_search(prompt)
if not content:
return render_template('index.html',
error="No relevant match data found.",
prompt=prompt)
# Generate answer
conversation = [
{"role": "system", "content": """You are a football analysis assistant specialized in Premier League matches.
Analyze the provided match data and give clear, concise insights about team performance, scores, and results.
Base your analysis only on the data provided."""},
{"role": "user", "content": f"""Based on the following football match data:
{content}
Question: {prompt}
Please analyze this data and provide insights about the matches, including:
- Match results and scores
- Team performance
- Notable statistics
Please base your analysis only on the data provided."""}
]
response = generate_answer(conversation)
return render_template('index.html',
prompt=prompt,
response=response,
dataset_info=references)
return render_template('index.html')
except Exception as e:
logger.error(f"Error in index route: {str(e)}")
return render_template('index.html',
error="An error occurred. Please try again.")
if __name__ == '__main__':
# Only for local development
application.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))
按照以下步骤将 Python 机器人应用程序部署到 Azure:
config.py
中更新 Microsoft 应用程序 ID 和密码。class DefaultConfig:
""" Bot Configuration """
PORT = 3978
APP_ID = os.environ.get("MicrosoftAppId", "Microsoft_APP_ID")
APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "<Microsoft_App_Password>")
用如下函数包围路线:
def init_func(argv):
APP = web.Application(middlewares=[aiohttp_error_middleware])
APP.router.add_post("/api/messages", messages)
return APP
if __name__ == "__main__":
APP = init_func(None)
try:
web.run_app(APP, host="0.0.0.0", port=CONFIG.PORT)
except Exception as error:
raise error
将 Web 应用程序部署到 Azure。
机器人配置:
使用
gunicorn
命令运行已部署的机器人。
如果您有如上所述的 init_func()
,请使用以下命令:
python3 -m aiohttp.web -H 0.0.0.0 -P 8000 app:init_func
Bot=>在网络聊天中测试:
导航到
App Service=>Monitoring
下的日志流并检查有关该问题的详细错误。
打开 Web 应用程序的 KUDU 站点,然后检查
Site/wwwroot
文件夹中是否存在应用程序文件夹。