AgentCoreService/history_dialogue_service.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

import requests
import logging
import json

import configs


class HistoryDialogueService:
    def __init__(self, base_url: str):
        self.base_url = base_url

    def get_dialogue_history(self, staff_id: str, user_id: str, max_count: int = 10):
        url = f"{self.base_url}?sender={staff_id}&receiver={user_id}"
        response = requests.post(url, headers={
            'Content-Type': 'application/json'
        })
        if response.status_code != 200:
            raise Exception("Request error [{}]: {}".format(response.status_code, response.text))
        data = response.json()
        if not data.get('success', False):
            raise Exception("Error in response: {}".format(data.get('message', 'no message returned')))
        data = data.get('data', [])
        ret = []
        for record in data:
            sender = record.get('sender')
            if sender == user_id:
                role = 'user'
            elif sender == staff_id:
                role = 'assistant'
            else:
                logging.warning("Unknown sender in dialogue history: {}".format(sender))
                continue
            ret.append({
                'role': role,
                'content': record.get('content', ''),
                'timestamp': record.get('sendtime', 0)
            })
        ret = sorted(ret, key=lambda x: x['timestamp'])
        return ret[-max_count:] if len(ret) > max_count else ret


if __name__ == '__main__':
    api_url = configs.get()['storage']['history_dialogue']['api_base_url']
    service = HistoryDialogueService(api_url)
    resp = service.get_dialogue_history(staff_id='1688854492669990', user_id='7881301263964433')
    print(resp)