Spaces:
Running
Running
File size: 5,845 Bytes
dde49a0 720c1cf c384631 9a577ee 1e2f57f 9ca54c4 aae4949 9ca54c4 1e2f57f 9ca54c4 aae4949 8c1f2d9 c384631 9ca54c4 c384631 9ca54c4 c384631 724f683 9ca54c4 c384631 67db7ba 9ca54c4 67db7ba c384631 9ca54c4 c384631 720c1cf aae4949 1111830 1e2f57f 9a577ee d5176a8 9a577ee 1e2f57f d5176a8 2436a6f 8c1f2d9 2436a6f 6642f5e 9a577ee 8a8fe1d 6642f5e c384631 9a577ee 2436a6f 9a577ee 8a8fe1d 6642f5e 9a577ee 1e2f57f 8a8fe1d d5176a8 1e2f57f 9a577ee 8a8fe1d 9a577ee 1e2f57f e340463 1e2f57f 8a8fe1d 2436a6f 1e2f57f 2436a6f 9a577ee d5176a8 1e2f57f 9ca54c4 1e2f57f 9ca54c4 1e2f57f c384631 1e2f57f 0efc2d5 1e2f57f 0efc2d5 1e2f57f 0efc2d5 1e2f57f 5684fd5 d5176a8 8a8fe1d 14dbe6a 9ca54c4 8a8fe1d 9ca54c4 1e2f57f c384631 8a8fe1d c384631 1e2f57f c384631 d5176a8 c07be57 |
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 185 186 |
import re
import time
import g4f
from datetime import datetime
from g4f import ChatCompletion
from flask import request, Response, stream_with_context
from requests import get
from server.config import special_instructions
class Backend_Api:
def __init__(self, bp, config: dict) -> None:
"""
Initialize the Backend_Api class.
:param app: Flask application instance
:param config: Configuration dictionary
"""
self.bp = bp
self.routes = {
'/backend-api/v2/conversation': {
'function': self._conversation,
'methods': ['POST']
}
}
def _conversation(self):
"""
Handles the conversation route.
:return: Response object containing the generated conversation stream
"""
max_retries = 3
retries = 0
conversation_id = request.json['conversation_id']
while retries < max_retries:
try:
jailbreak = request.json['jailbreak']
model = request.json['model']
messages = build_messages(jailbreak)
# Generate response
response = ChatCompletion.create(
model=model,
stream=True,
chatId=conversation_id,
messages=messages
)
return Response(stream_with_context(generate_stream(response, jailbreak)), mimetype='text/event-stream')
except Exception as e:
print(e)
print(e.__traceback__.tb_next)
retries += 1
if retries >= max_retries:
return {
'_action': '_ask',
'success': False,
"error": f"an error occurred {str(e)}"
}, 400
time.sleep(3) # Wait 3 second before trying again
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]
# Add the existing conversation
conversation = _conversation
# Add web results if enabled
if internet_access:
current_date = datetime.now().strftime("%Y-%m-%d")
query = f'Current date: {current_date}. ' + prompt["content"]
search_results = fetch_search_results(query)
conversation.extend(search_results)
# Add jailbreak instructions if enabled
if jailbreak_instructions := getJailbreak(jailbreak):
conversation.extend(jailbreak_instructions)
# Add the prompt
conversation.append(prompt)
# Reduce conversation size to avoid API Token quantity error
if len(conversation) > 3:
conversation = conversation[-4:]
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,
})
snippets = ""
for index, result in enumerate(search.json()):
snippet = f'[{index + 1}] "{result["snippet"]}" URL:{result["link"]}.'
snippets += snippet
response = "Here are some updated web searches. Use this to improve user response:"
response += snippets
return [{'role': 'system', 'content': response}]
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 getJailbreak(jailbreak):
response_jailbreak = ''
jailbroken_checked = False
for message in response:
response_jailbreak += message
if jailbroken_checked:
yield message
else:
if response_jailbroken_success(response_jailbreak):
jailbroken_checked = True
if response_jailbroken_failed(response_jailbreak):
yield response_jailbreak
jailbroken_checked = True
else:
yield from response
def response_jailbroken_success(response: str) -> bool:
"""Check if the response has been jailbroken.
:param response: Response string
:return: Boolean indicating if the response has been jailbroken
"""
act_match = re.search(r'ACT:', response, flags=re.DOTALL)
return bool(act_match)
def response_jailbroken_failed(response):
"""
Check if the response has not been jailbroken.
:param response: Response string
:return: Boolean indicating if the response has not been jailbroken
"""
return False if len(response) < 4 else not (response.startswith("GPT:") or response.startswith("ACT:"))
def getJailbreak(jailbreak):
"""
Check if jailbreak instructions are provided.
:param jailbreak: Jailbreak instruction string
:return: Jailbreak instructions if provided, otherwise None
"""
if jailbreak != "default":
special_instructions[jailbreak][0]['content'] += special_instructions['two_responses_instruction']
if jailbreak in special_instructions:
special_instructions[jailbreak]
return special_instructions[jailbreak]
else:
return None
else:
return None
|