File size: 5,652 Bytes
70422d8
dde49a0
1e2f57f
 
1111830
aae4949
 
 
1e2f57f
aae4949
 
1e2f57f
 
aae4949
 
 
1e2f57f
 
 
 
 
 
aae4949
70422d8
aae4949
 
 
 
 
 
 
70422d8
 
 
 
 
aae4949
1e2f57f
 
 
 
 
aae4949
 
8a8fe1d
1e2f57f
aae4949
1e2f57f
 
 
aae4949
1e2f57f
aae4949
 
 
 
 
 
 
 
 
1111830
 
1e2f57f
 
 
d5176a8
1e2f57f
 
 
 
 
 
d5176a8
1e2f57f
 
 
 
 
 
 
 
1111830
1e2f57f
 
 
 
1111830
1e2f57f
 
8a8fe1d
1e2f57f
 
8a8fe1d
1e2f57f
 
 
 
8a8fe1d
1e2f57f
 
 
 
8a8fe1d
d5176a8
1e2f57f
 
 
8a8fe1d
1e2f57f
 
 
 
 
 
 
 
8a8fe1d
1e2f57f
 
 
 
d5176a8
1e2f57f
d5176a8
 
1e2f57f
 
 
 
 
 
 
 
d5176a8
1e2f57f
 
 
 
 
 
 
 
 
 
 
d5176a8
8a8fe1d
1e2f57f
 
 
8a8fe1d
1e2f57f
 
 
 
 
 
8a8fe1d
 
1e2f57f
 
 
 
 
 
 
 
 
 
d5176a8
 
 
1e2f57f
 
 
 
 
 
d5176a8
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
import threading
import re
import g4f
from g4f import ChatCompletion
from googletrans import Translator
from flask import request
from datetime import datetime
from requests import get
from server.auto_proxy import get_random_proxy, update_working_proxies
from server.config import special_instructions

# Backend_Api class definition


class Backend_Api:
    def __init__(self, app, config: dict) -> None:
        """  
        Initialize the Backend_Api class.  

        :param app: Flask application instance  
        :param config: Configuration dictionary  
        """
        self.app = app
        self.use_auto_proxy = config['use_auto_proxy']
        self.routes = {
            '/backend-api/v2/conversation': {
                'function': self._conversation,
                'methods': ['POST']
            }
        }

        if self.use_auto_proxy:
            update_proxies = threading.Thread(
                target=update_working_proxies, daemon=True)
            update_proxies.start()

    def _conversation(self):
        """  
        Handles the conversation route.  

        :return: Response object containing the generated conversation stream  
        """
        try:
            jailbreak = request.json['jailbreak']
            model = request.json['model']
            messages = build_messages(jailbreak)

            # Generate response
            response = ChatCompletion.create(model=model, stream=True,
                                             messages=messages, provider=g4f.Provider.Forefront)

            return self.app.response_class(generate_stream(response, jailbreak), mimetype='text/event-stream')

        except Exception as e:
            print(e)
            print(e.__traceback__.tb_next)
            return {
                '_action': '_ask',
                'success': False,
                "error": f"an error occurred {str(e)}"
            }, 400


def build_messages(jailbreak):
    """  
    Build the messages for the conversation.  

    :param jailbreak: Jailbreak instruction string  
    :return: List of messages for the conversation  
    """
    _conversation = request.json['meta']['content']['conversation']
    internet_access = request.json['meta']['content']['internet_access']
    prompt = request.json['meta']['content']['parts'][0]

    # Generate system message
    current_date = datetime.now().strftime("%Y-%m-%d")
    system_message = (
        f'You are ChatGPT also known as ChatGPT, a large language model trained by OpenAI. '
        f'Strictly follow the users instructions. '
        f'Knowledge cutoff: 2021-09-01 Current date: {current_date}. '
        f'{set_response_language(prompt)}'
    )

    # Fetch search results if internet access is allowed
    extra = []
    if internet_access:
        extra = fetch_search_results(prompt["content"])

    # Initialize the conversation with the system message
    conversation = [{'role': 'system', 'content': system_message}]

    # Add extra results
    conversation += extra

    # Add jailbreak instructions, if any
    jailbreak_instructions = isJailbreak(jailbreak)
    if jailbreak_instructions:
        conversation += jailbreak_instructions

    # Add the existing conversation and the prompt
    conversation += [prompt]

    return conversation


def fetch_search_results(query):
    """  
    Fetch search results for a given query.  

    :param query: Search query string  
    :return: List of search results  
    """
    search = get('https://ddg-api.herokuapp.com/search',
                 params={
                     'query': query,
                     'limit': 3,
                 })

    results = []
    for index, result in enumerate(search.json()):
        snippet = f'[{index}] "{result["snippet"]}"\nURL:{result["link"]}\n\n'
        results.append({'role': 'user', 'content': snippet})

    return results


def generate_stream(response, jailbreak):
    """  
    Generate the conversation stream.  

    :param response: Response object from ChatCompletion.create  
    :param jailbreak: Jailbreak instruction string  
    :return: Generator object yielding messages in the conversation  
    """
    if isJailbreak(jailbreak):
        response_jailbreak = ''
        unlocked = False
        for message in response:
            response_jailbreak += message
            if unlocked:
                yield message
            if response_jailbroken(response_jailbreak):
                unlocked = True
    else:
        for message in response:
            yield message


def response_jailbroken(response):
    """  
    Check if the response has been jailbroken.  

    :param response: Response string  
    :return: Boolean indicating if the response has been jailbroken  
    """
    act_pattern = re.compile(r'ACT:', flags=re.DOTALL)
    act_match = act_pattern.search(response)
    return act_match


def set_response_language(prompt):
    """  
    Set the response language based on the prompt content.  

    :param prompt: Prompt dictionary  
    :return: String indicating the language to be used for the response  
    """
    translator = Translator()
    detected_language = translator.detect(prompt['content']).lang
    return f"You will respond in the language: {detected_language}. "


def isJailbreak(jailbreak):
    """  
    Check if jailbreak instructions are provided.  

    :param jailbreak: Jailbreak instruction string  
    :return: Jailbreak instructions if provided, otherwise None  
    """
    if jailbreak != "Default":
        return special_instructions[jailbreak] if jailbreak in special_instructions else None
    else:
        return None