backend.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  1. import g4f
  2. from g4f.Provider import __providers__
  3. import json
  4. from flask import request, Flask
  5. from .internet import get_search_message
  6. from g4f import debug
  7. debug.logging = True
  8. class Backend_Api:
  9. def __init__(self, app: Flask) -> None:
  10. self.app: Flask = app
  11. self.routes = {
  12. '/backend-api/v2/models': {
  13. 'function': self.models,
  14. 'methods' : ['GET']
  15. },
  16. '/backend-api/v2/providers': {
  17. 'function': self.providers,
  18. 'methods' : ['GET']
  19. },
  20. '/backend-api/v2/version': {
  21. 'function': self.version,
  22. 'methods' : ['GET']
  23. },
  24. '/backend-api/v2/conversation': {
  25. 'function': self._conversation,
  26. 'methods': ['POST']
  27. },
  28. '/backend-api/v2/gen.set.summarize:title': {
  29. 'function': self._gen_title,
  30. 'methods': ['POST']
  31. },
  32. '/backend-api/v2/error': {
  33. 'function': self.error,
  34. 'methods': ['POST']
  35. }
  36. }
  37. def error(self):
  38. print(request.json)
  39. return 'ok', 200
  40. def models(self):
  41. return g4f._all_models
  42. def providers(self):
  43. return [
  44. provider.__name__ for provider in __providers__ if provider.working
  45. ]
  46. def version(self):
  47. return {
  48. "version": debug.get_version(),
  49. "lastet_version": debug.get_lastet_version(),
  50. }
  51. def _gen_title(self):
  52. return {
  53. 'title': ''
  54. }
  55. def _conversation(self):
  56. #jailbreak = request.json['jailbreak']
  57. messages = request.json['meta']['content']['parts']
  58. if request.json['internet_access']:
  59. messages[-1]["content"] = get_search_message(messages[-1]["content"])
  60. model = request.json.get('model')
  61. model = model if model else g4f.models.default
  62. provider = request.json.get('provider', '').replace('g4f.Provider.', '')
  63. provider = provider if provider and provider != "Auto" else None
  64. def try_response():
  65. try:
  66. yield from g4f.ChatCompletion.create(
  67. model=model,
  68. provider=provider,
  69. messages=messages,
  70. stream=True,
  71. ignore_stream_and_auth=True
  72. )
  73. except Exception as e:
  74. print(e)
  75. yield json.dumps({
  76. 'code' : 'G4F_ERROR',
  77. '_action': '_ask',
  78. 'success': False,
  79. 'error' : f'{e.__class__.__name__}: {e}'
  80. })
  81. return self.app.response_class(try_response(), mimetype='text/event-stream')