File size: 3,600 Bytes
a4b70d9 |
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 |
import sys, re
from pathlib import Path
from os import path
sys.path.append(str(Path(__file__).parent.parent.parent))
# Enable logging
import g4f
g4f.debug.logging = True
# Read auth files
from g4f.cookies import read_cookie_files
read_cookie_files()
def read_code(text):
if match := re.search(r"```(python|py|)\n(?P<code>[\S\s]+?)\n```", text):
return match.group("code")
return text
def input_command():
print("Enter/Paste the cURL command. Ctrl-D or Ctrl-Z ( windows ) to save it.")
contents = []
while True:
try:
line = input()
except EOFError:
break
contents.append(line)
return "\n".join(contents)
name = input("Name: ")
provider_path = f"g4f/Provider/{name}.py"
example = """
from __future__ import annotations
from aiohttp import ClientSession
from ..typing import AsyncResult, Messages
from .base_provider import AsyncGeneratorProvider, ProviderModelMixin
from .helper import format_prompt
class {name}(AsyncGeneratorProvider, ProviderModelMixin):
label = ""
url = "https://example.com"
api_endpoint = "https://example.com/api/completion"
working = True
needs_auth = False
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = ''
models = ['', '']
model_aliases = {
"alias1": "model1",
}
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
**kwargs
) -> AsyncResult:
model = cls.get_model(model)
headers = {{
"authority": "example.com",
"accept": "application/json",
"origin": cls.url,
"referer": f"{{cls.url}}/chat",
}}
async with ClientSession(headers=headers) as session:
prompt = format_prompt(messages)
data = {{
"prompt": prompt,
"model": model,
}}
async with session.post(f"{{cls.url}}/api/chat", json=data, proxy=proxy) as response:
response.raise_for_status()
async for chunk in response.content:
if chunk:
yield chunk.decode()
"""
if not path.isfile(provider_path):
command = input_command()
prompt = f"""
Create a provider from a cURL command. The command is:
```bash
{command}
```
A example for a provider:
```python
{example}
```
The name for the provider class:
{name}
Replace "hello" with `format_prompt(messages)`.
And replace "gpt-3.5-turbo" with `model`.
"""
print("Create code...")
response = []
for chunk in g4f.ChatCompletion.create(
model=g4f.models.default,
messages=prompt,
stream=True,
):
print(chunk, end="", flush=True)
if not isinstance(chunk, Exception):
response.append(str(chunk))
print()
response = "".join(response)
if code := read_code(response):
with open(provider_path, "w") as file:
file.write(code)
print("Saved at:", provider_path)
with open("g4f/Provider/__init__.py", "a") as file:
file.write(f"\nfrom .{name} import {name}")
else:
with open(provider_path, "r") as file:
code = file.read()
|