telegram_bot_aws_lambda

lambda.py

import os
import json
import logging
from urllib import request
from datetime import datetime, timezone
from dataclasses import dataclass

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("interactions")  # Replace with your DynamoDB table name


@dataclass(kw_only=True, frozen=True)
class Interaction:
    user: str
    system: str

logger = logging.getLogger()
logger.setLevel(logging.INFO)

telegram_token = os.environ["TELEGRAM_TOKEN"]
mistral_token = os.environ["MISTRAL_TOKEN"]

headers = {
    "Authorization": f"Bearer {mistral_token}",
    "User-Agent": "python-requests/2.31.0",
    "Content-Type": "application/json",
}


def save_interaction(chat_id, message, response):
    timestamp = datetime.now(timezone.utc).isoformat()
    table.put_item(
        Item={
            "chat_id": str(chat_id),
            "timestamp": timestamp,
            "message": message,
            "response": response,
        }
    )
    
def get_last_interactions(chat_id):
    response = table.query(
        KeyConditionExpression=Key('chat_id').eq(str(chat_id)),
        ScanIndexForward=False,
        Limit=5
    )
    items =  response.get('Items', [])
    return [Interaction(user=it['message'], system=it['response']) for it in reversed(items)]


def chat(msg):
    body = {
        "model": "mistral-medium",
        "messages": [{"role": "user", "content": msg}],
        "temperature": 0.7,
        "top_p": 1,
        "max_tokens": 8000,
        "stream": False,
        "safe_mode": False,
        "random_seed": None,
    }

    data = json.dumps(body).encode()
    req = request.Request(
        "https://api.mistral.ai/v1/chat/completions", data=data, headers=headers
    )

    try:
        with request.urlopen(req) as res:
            response_body = res.read()
            response_data = json.loads(response_body)
            return response_data["choices"][0]["message"]["content"]
    except Exception as e:
        print(f"An error occurred: {e}")
        return None


def lambda_handler(event, context):
    msg = json.loads(event["body"])

    chat_id = msg.get("message", {}).get("chat", {}).get("id")
    message = msg.get("message", {}).get("text")
    if not chat_id or not message:
        return {"statusCode": 200}
    history = '\n'.join(f"user: {i.user}\nsystem: {i.system}" for i in get_last_interactions(chat_id))
    response = chat(f"{history}\n{message}")

    if not response:
        return {"statusCode": 200}

    msg = {
        "chat_id": chat_id,
        "text": response,
    }

    data = json.dumps(msg).encode()
    url = f"https://api.telegram.org/bot{telegram_token}/sendMessage"
    req = request.Request(url, data=data, headers={"Content-Type": "application/json"})

    request.urlopen(req)
    save_interaction(chat_id, message, response)
    return {"statusCode": 200}