task.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201
  1. import math
  2. import re
  3. from datetime import datetime
  4. from typing import Optional
  5. from open_webui.utils.misc import get_last_user_message, get_messages_content
  6. def prompt_template(
  7. template: str, user_name: Optional[str] = None, user_location: Optional[str] = None
  8. ) -> str:
  9. # Get the current date
  10. current_date = datetime.now()
  11. # Format the date to YYYY-MM-DD
  12. formatted_date = current_date.strftime("%Y-%m-%d")
  13. formatted_time = current_date.strftime("%I:%M:%S %p")
  14. template = template.replace("{{CURRENT_DATE}}", formatted_date)
  15. template = template.replace("{{CURRENT_TIME}}", formatted_time)
  16. template = template.replace(
  17. "{{CURRENT_DATETIME}}", f"{formatted_date} {formatted_time}"
  18. )
  19. if user_name:
  20. # Replace {{USER_NAME}} in the template with the user's name
  21. template = template.replace("{{USER_NAME}}", user_name)
  22. else:
  23. # Replace {{USER_NAME}} in the template with "Unknown"
  24. template = template.replace("{{USER_NAME}}", "Unknown")
  25. if user_location:
  26. # Replace {{USER_LOCATION}} in the template with the current location
  27. template = template.replace("{{USER_LOCATION}}", user_location)
  28. else:
  29. # Replace {{USER_LOCATION}} in the template with "Unknown"
  30. template = template.replace("{{USER_LOCATION}}", "Unknown")
  31. return template
  32. def replace_prompt_variable(template: str, prompt: str) -> str:
  33. def replacement_function(match):
  34. full_match = match.group(0)
  35. start_length = match.group(1)
  36. end_length = match.group(2)
  37. middle_length = match.group(3)
  38. if full_match == "{{prompt}}":
  39. return prompt
  40. elif start_length is not None:
  41. return prompt[: int(start_length)]
  42. elif end_length is not None:
  43. return prompt[-int(end_length) :]
  44. elif middle_length is not None:
  45. middle_length = int(middle_length)
  46. if len(prompt) <= middle_length:
  47. return prompt
  48. start = prompt[: math.ceil(middle_length / 2)]
  49. end = prompt[-math.floor(middle_length / 2) :]
  50. return f"{start}...{end}"
  51. return ""
  52. template = re.sub(
  53. r"{{prompt}}|{{prompt:start:(\d+)}}|{{prompt:end:(\d+)}}|{{prompt:middletruncate:(\d+)}}",
  54. replacement_function,
  55. template,
  56. )
  57. return template
  58. def title_generation_template(
  59. template: str, prompt: str, user: Optional[dict] = None
  60. ) -> str:
  61. template = replace_prompt_variable(template, prompt)
  62. template = prompt_template(
  63. template,
  64. **(
  65. {"user_name": user.get("name"), "user_location": user.get("location")}
  66. if user
  67. else {}
  68. ),
  69. )
  70. return template
  71. def replace_messages_variable(template: str, messages: list[str]) -> str:
  72. def replacement_function(match):
  73. full_match = match.group(0)
  74. start_length = match.group(1)
  75. end_length = match.group(2)
  76. middle_length = match.group(3)
  77. # Process messages based on the number of messages required
  78. if full_match == "{{MESSAGES}}":
  79. return get_messages_content(messages)
  80. elif start_length is not None:
  81. return get_messages_content(messages[: int(start_length)])
  82. elif end_length is not None:
  83. return get_messages_content(messages[-int(end_length) :])
  84. elif middle_length is not None:
  85. mid = int(middle_length)
  86. if len(messages) <= mid:
  87. return get_messages_content(messages)
  88. # Handle middle truncation: split to get start and end portions of the messages list
  89. half = mid // 2
  90. start_msgs = messages[:half]
  91. end_msgs = messages[-half:] if mid % 2 == 0 else messages[-(half + 1) :]
  92. formatted_start = get_messages_content(start_msgs)
  93. formatted_end = get_messages_content(end_msgs)
  94. return f"{formatted_start}\n{formatted_end}"
  95. return ""
  96. template = re.sub(
  97. r"{{MESSAGES}}|{{MESSAGES:START:(\d+)}}|{{MESSAGES:END:(\d+)}}|{{MESSAGES:MIDDLETRUNCATE:(\d+)}}",
  98. replacement_function,
  99. template,
  100. )
  101. return template
  102. def tags_generation_template(
  103. template: str, messages: list[dict], user: Optional[dict] = None
  104. ) -> str:
  105. prompt = get_last_user_message(messages)
  106. template = replace_prompt_variable(template, prompt)
  107. template = replace_messages_variable(template, messages)
  108. template = prompt_template(
  109. template,
  110. **(
  111. {"user_name": user.get("name"), "user_location": user.get("location")}
  112. if user
  113. else {}
  114. ),
  115. )
  116. return template
  117. def search_query_generation_template(
  118. template: str, messages: list[dict], user: Optional[dict] = None
  119. ) -> str:
  120. prompt = get_last_user_message(messages)
  121. template = replace_prompt_variable(template, prompt)
  122. template = replace_messages_variable(template, messages)
  123. template = prompt_template(
  124. template,
  125. **(
  126. {"user_name": user.get("name"), "user_location": user.get("location")}
  127. if user
  128. else {}
  129. ),
  130. )
  131. return template
  132. def moa_response_generation_template(
  133. template: str, prompt: str, responses: list[str]
  134. ) -> str:
  135. def replacement_function(match):
  136. full_match = match.group(0)
  137. start_length = match.group(1)
  138. end_length = match.group(2)
  139. middle_length = match.group(3)
  140. if full_match == "{{prompt}}":
  141. return prompt
  142. elif start_length is not None:
  143. return prompt[: int(start_length)]
  144. elif end_length is not None:
  145. return prompt[-int(end_length) :]
  146. elif middle_length is not None:
  147. middle_length = int(middle_length)
  148. if len(prompt) <= middle_length:
  149. return prompt
  150. start = prompt[: math.ceil(middle_length / 2)]
  151. end = prompt[-math.floor(middle_length / 2) :]
  152. return f"{start}...{end}"
  153. return ""
  154. template = re.sub(
  155. r"{{prompt}}|{{prompt:start:(\d+)}}|{{prompt:end:(\d+)}}|{{prompt:middletruncate:(\d+)}}",
  156. replacement_function,
  157. template,
  158. )
  159. responses = [f'"""{response}"""' for response in responses]
  160. responses = "\n\n".join(responses)
  161. template = template.replace("{{responses}}", responses)
  162. return template
  163. def tools_function_calling_generation_template(template: str, tools_specs: str) -> str:
  164. template = template.replace("{{TOOLS}}", tools_specs)
  165. return template