Programmiererforum

Code: Select all

import os
import sys
import json
import traceback
from dataclasses import asdict

from botbuilder.core import MemoryStorage, TurnContext
from botbuilder.schema import TextFormatTypes
from teams import Application, ApplicationOptions, TeamsAdapter
from teams.ai import AIOptions
from teams.ai.models import AzureOpenAIModelOptions, OpenAIModel, OpenAIModelOptions
from teams.ai.planners import ActionPlanner, ActionPlannerOptions
from teams.ai.prompts import PromptManager, PromptManagerOptions
from teams.state import TurnState
from teams.feedback_loop_data import FeedbackLoopData

from config import Config

config = Config()

# Create AI components
model: OpenAIModel

model = OpenAIModel(
AzureOpenAIModelOptions(
api_key=config.AZURE_OPENAI_API_KEY,
default_model=config.AZURE_OPENAI_MODEL_DEPLOYMENT_NAME,
endpoint=config.AZURE_OPENAI_ENDPOINT,
)
)

prompts = PromptManager(PromptManagerOptions(prompts_folder=f"{os.getcwd()}/prompts"))

planner = ActionPlanner(
ActionPlannerOptions(model=model, prompts=prompts, default_prompt="chat")
)

# Define storage and application
storage = MemoryStorage()
bot_app = Application[TurnState](
ApplicationOptions(
bot_app_id=config.APP_ID,
storage=storage,
adapter=TeamsAdapter(config),
ai=AIOptions(planner=planner, enable_feedback_loop=True),
)
)

# Utility function to send Markdown-formatted messages
async def send_markdown(context: TurnContext, text: str):
await context.send_activity(
text=text,
text_format="markdown"
)

@bot_app.error
async def on_error(context: TurnContext, error: Exception):
# Log the error
print(f"\n\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
traceback.print_exc()

# Send a Markdown-formatted error message to the user
await send_markdown(context, "**Oops!** The bot encountered an error or bug.\n\nPlease try again later.")

@bot_app.feedback_loop()
async def feedback_loop(_context: TurnContext, _state: TurnState, feedback_loop_data: FeedbackLoopData):
# Format feedback data with proper line breaks
json_content = json.dumps(asdict(feedback_loop_data), indent=4)
feedback_text = (
"Your feedback:\n\n"
"```  "  # Start of preformatted block
f"{json_content}\n"  # Insert formatted JSON
"```"  # End of preformatted block
)
await send_markdown(_context, feedback_text)

# Add any additional response handlers and ensure they use send_markdown
async def handle_user_message(context: TurnContext):
user_message = context.activity.text
response = f"**You said:** {user_message}\n\nHow can I assist you further?"
await send_markdown(context, response)