middleware.py 74 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952
  1. import time
  2. import logging
  3. import sys
  4. import os
  5. import base64
  6. import asyncio
  7. from aiocache import cached
  8. from typing import Any, Optional
  9. import random
  10. import json
  11. import html
  12. import inspect
  13. import re
  14. import ast
  15. from uuid import uuid4
  16. from concurrent.futures import ThreadPoolExecutor
  17. from fastapi import Request
  18. from fastapi import BackgroundTasks
  19. from starlette.responses import Response, StreamingResponse
  20. from open_webui.models.chats import Chats
  21. from open_webui.models.users import Users
  22. from open_webui.socket.main import (
  23. get_event_call,
  24. get_event_emitter,
  25. get_active_status_by_user_id,
  26. )
  27. from open_webui.routers.tasks import (
  28. generate_queries,
  29. generate_title,
  30. generate_image_prompt,
  31. generate_chat_tags,
  32. )
  33. from open_webui.routers.retrieval import process_web_search, SearchForm
  34. from open_webui.routers.images import image_generations, GenerateImageForm
  35. from open_webui.routers.pipelines import (
  36. process_pipeline_inlet_filter,
  37. process_pipeline_outlet_filter,
  38. )
  39. from open_webui.utils.webhook import post_webhook
  40. from open_webui.models.users import UserModel
  41. from open_webui.models.functions import Functions
  42. from open_webui.models.models import Models
  43. from open_webui.retrieval.utils import get_sources_from_files
  44. from open_webui.utils.chat import generate_chat_completion
  45. from open_webui.utils.task import (
  46. get_task_model_id,
  47. rag_template,
  48. tools_function_calling_generation_template,
  49. )
  50. from open_webui.utils.misc import (
  51. deep_update,
  52. get_message_list,
  53. add_or_update_system_message,
  54. add_or_update_user_message,
  55. get_last_user_message,
  56. get_last_assistant_message,
  57. prepend_to_first_user_message_content,
  58. )
  59. from open_webui.utils.tools import get_tools
  60. from open_webui.utils.plugin import load_function_module_by_id
  61. from open_webui.utils.filter import (
  62. get_sorted_filter_ids,
  63. process_filter_functions,
  64. )
  65. from open_webui.utils.code_interpreter import execute_code_jupyter
  66. from open_webui.tasks import create_task
  67. from open_webui.config import (
  68. CACHE_DIR,
  69. DEFAULT_TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE,
  70. DEFAULT_CODE_INTERPRETER_PROMPT,
  71. )
  72. from open_webui.env import (
  73. SRC_LOG_LEVELS,
  74. GLOBAL_LOG_LEVEL,
  75. BYPASS_MODEL_ACCESS_CONTROL,
  76. ENABLE_REALTIME_CHAT_SAVE,
  77. )
  78. from open_webui.constants import TASKS
  79. logging.basicConfig(stream=sys.stdout, level=GLOBAL_LOG_LEVEL)
  80. log = logging.getLogger(__name__)
  81. log.setLevel(SRC_LOG_LEVELS["MAIN"])
  82. async def chat_completion_tools_handler(
  83. request: Request, body: dict, user: UserModel, models, tools
  84. ) -> tuple[dict, dict]:
  85. async def get_content_from_response(response) -> Optional[str]:
  86. content = None
  87. if hasattr(response, "body_iterator"):
  88. async for chunk in response.body_iterator:
  89. data = json.loads(chunk.decode("utf-8"))
  90. content = data["choices"][0]["message"]["content"]
  91. # Cleanup any remaining background tasks if necessary
  92. if response.background is not None:
  93. await response.background()
  94. else:
  95. content = response["choices"][0]["message"]["content"]
  96. return content
  97. def get_tools_function_calling_payload(messages, task_model_id, content):
  98. user_message = get_last_user_message(messages)
  99. history = "\n".join(
  100. f"{message['role'].upper()}: \"\"\"{message['content']}\"\"\""
  101. for message in messages[::-1][:4]
  102. )
  103. prompt = f"History:\n{history}\nQuery: {user_message}"
  104. return {
  105. "model": task_model_id,
  106. "messages": [
  107. {"role": "system", "content": content},
  108. {"role": "user", "content": f"Query: {prompt}"},
  109. ],
  110. "stream": False,
  111. "metadata": {"task": str(TASKS.FUNCTION_CALLING)},
  112. }
  113. task_model_id = get_task_model_id(
  114. body["model"],
  115. request.app.state.config.TASK_MODEL,
  116. request.app.state.config.TASK_MODEL_EXTERNAL,
  117. models,
  118. )
  119. skip_files = False
  120. sources = []
  121. specs = [tool["spec"] for tool in tools.values()]
  122. tools_specs = json.dumps(specs)
  123. if request.app.state.config.TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE != "":
  124. template = request.app.state.config.TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE
  125. else:
  126. template = DEFAULT_TOOLS_FUNCTION_CALLING_PROMPT_TEMPLATE
  127. tools_function_calling_prompt = tools_function_calling_generation_template(
  128. template, tools_specs
  129. )
  130. log.info(f"{tools_function_calling_prompt=}")
  131. payload = get_tools_function_calling_payload(
  132. body["messages"], task_model_id, tools_function_calling_prompt
  133. )
  134. try:
  135. response = await generate_chat_completion(request, form_data=payload, user=user)
  136. log.debug(f"{response=}")
  137. content = await get_content_from_response(response)
  138. log.debug(f"{content=}")
  139. if not content:
  140. return body, {}
  141. try:
  142. content = content[content.find("{") : content.rfind("}") + 1]
  143. if not content:
  144. raise Exception("No JSON object found in the response")
  145. result = json.loads(content)
  146. async def tool_call_handler(tool_call):
  147. nonlocal skip_files
  148. log.debug(f"{tool_call=}")
  149. tool_function_name = tool_call.get("name", None)
  150. if tool_function_name not in tools:
  151. return body, {}
  152. tool_function_params = tool_call.get("parameters", {})
  153. try:
  154. required_params = (
  155. tools[tool_function_name]
  156. .get("spec", {})
  157. .get("parameters", {})
  158. .get("required", [])
  159. )
  160. tool_function = tools[tool_function_name]["callable"]
  161. tool_function_params = {
  162. k: v
  163. for k, v in tool_function_params.items()
  164. if k in required_params
  165. }
  166. tool_output = await tool_function(**tool_function_params)
  167. except Exception as e:
  168. tool_output = str(e)
  169. if isinstance(tool_output, str):
  170. if tools[tool_function_name]["citation"]:
  171. sources.append(
  172. {
  173. "source": {
  174. "name": f"TOOL:{tools[tool_function_name]['toolkit_id']}/{tool_function_name}"
  175. },
  176. "document": [tool_output],
  177. "metadata": [
  178. {
  179. "source": f"TOOL:{tools[tool_function_name]['toolkit_id']}/{tool_function_name}"
  180. }
  181. ],
  182. }
  183. )
  184. else:
  185. sources.append(
  186. {
  187. "source": {},
  188. "document": [tool_output],
  189. "metadata": [
  190. {
  191. "source": f"TOOL:{tools[tool_function_name]['toolkit_id']}/{tool_function_name}"
  192. }
  193. ],
  194. }
  195. )
  196. if tools[tool_function_name]["file_handler"]:
  197. skip_files = True
  198. # check if "tool_calls" in result
  199. if result.get("tool_calls"):
  200. for tool_call in result.get("tool_calls"):
  201. await tool_call_handler(tool_call)
  202. else:
  203. await tool_call_handler(result)
  204. except Exception as e:
  205. log.exception(f"Error: {e}")
  206. content = None
  207. except Exception as e:
  208. log.exception(f"Error: {e}")
  209. content = None
  210. log.debug(f"tool_contexts: {sources}")
  211. if skip_files and "files" in body.get("metadata", {}):
  212. del body["metadata"]["files"]
  213. return body, {"sources": sources}
  214. async def chat_web_search_handler(
  215. request: Request, form_data: dict, extra_params: dict, user
  216. ):
  217. event_emitter = extra_params["__event_emitter__"]
  218. await event_emitter(
  219. {
  220. "type": "status",
  221. "data": {
  222. "action": "web_search",
  223. "description": "Generating search query",
  224. "done": False,
  225. },
  226. }
  227. )
  228. messages = form_data["messages"]
  229. user_message = get_last_user_message(messages)
  230. queries = []
  231. try:
  232. res = await generate_queries(
  233. request,
  234. {
  235. "model": form_data["model"],
  236. "messages": messages,
  237. "prompt": user_message,
  238. "type": "web_search",
  239. },
  240. user,
  241. )
  242. response = res["choices"][0]["message"]["content"]
  243. try:
  244. bracket_start = response.find("{")
  245. bracket_end = response.rfind("}") + 1
  246. if bracket_start == -1 or bracket_end == -1:
  247. raise Exception("No JSON object found in the response")
  248. response = response[bracket_start:bracket_end]
  249. queries = json.loads(response)
  250. queries = queries.get("queries", [])
  251. except Exception as e:
  252. queries = [response]
  253. except Exception as e:
  254. log.exception(e)
  255. queries = [user_message]
  256. if len(queries) == 0:
  257. await event_emitter(
  258. {
  259. "type": "status",
  260. "data": {
  261. "action": "web_search",
  262. "description": "No search query generated",
  263. "done": True,
  264. },
  265. }
  266. )
  267. return form_data
  268. searchQuery = queries[0]
  269. await event_emitter(
  270. {
  271. "type": "status",
  272. "data": {
  273. "action": "web_search",
  274. "description": 'Searching "{{searchQuery}}"',
  275. "query": searchQuery,
  276. "done": False,
  277. },
  278. }
  279. )
  280. try:
  281. results = await process_web_search(
  282. request,
  283. SearchForm(
  284. **{
  285. "query": searchQuery,
  286. }
  287. ),
  288. user,
  289. )
  290. if results:
  291. await event_emitter(
  292. {
  293. "type": "status",
  294. "data": {
  295. "action": "web_search",
  296. "description": "Searched {{count}} sites",
  297. "query": searchQuery,
  298. "urls": results["filenames"],
  299. "done": True,
  300. },
  301. }
  302. )
  303. files = form_data.get("files", [])
  304. if request.app.state.config.RAG_WEB_SEARCH_FULL_CONTEXT:
  305. files.append(
  306. {
  307. "docs": results.get("docs", []),
  308. "name": searchQuery,
  309. "type": "web_search_docs",
  310. "urls": results["filenames"],
  311. }
  312. )
  313. else:
  314. files.append(
  315. {
  316. "collection_name": results["collection_name"],
  317. "name": searchQuery,
  318. "type": "web_search_results",
  319. "urls": results["filenames"],
  320. }
  321. )
  322. form_data["files"] = files
  323. else:
  324. await event_emitter(
  325. {
  326. "type": "status",
  327. "data": {
  328. "action": "web_search",
  329. "description": "No search results found",
  330. "query": searchQuery,
  331. "done": True,
  332. "error": True,
  333. },
  334. }
  335. )
  336. except Exception as e:
  337. log.exception(e)
  338. await event_emitter(
  339. {
  340. "type": "status",
  341. "data": {
  342. "action": "web_search",
  343. "description": 'Error searching "{{searchQuery}}"',
  344. "query": searchQuery,
  345. "done": True,
  346. "error": True,
  347. },
  348. }
  349. )
  350. return form_data
  351. async def chat_image_generation_handler(
  352. request: Request, form_data: dict, extra_params: dict, user
  353. ):
  354. __event_emitter__ = extra_params["__event_emitter__"]
  355. await __event_emitter__(
  356. {
  357. "type": "status",
  358. "data": {"description": "Generating an image", "done": False},
  359. }
  360. )
  361. messages = form_data["messages"]
  362. user_message = get_last_user_message(messages)
  363. prompt = user_message
  364. negative_prompt = ""
  365. if request.app.state.config.ENABLE_IMAGE_PROMPT_GENERATION:
  366. try:
  367. res = await generate_image_prompt(
  368. request,
  369. {
  370. "model": form_data["model"],
  371. "messages": messages,
  372. },
  373. user,
  374. )
  375. response = res["choices"][0]["message"]["content"]
  376. try:
  377. bracket_start = response.find("{")
  378. bracket_end = response.rfind("}") + 1
  379. if bracket_start == -1 or bracket_end == -1:
  380. raise Exception("No JSON object found in the response")
  381. response = response[bracket_start:bracket_end]
  382. response = json.loads(response)
  383. prompt = response.get("prompt", [])
  384. except Exception as e:
  385. prompt = user_message
  386. except Exception as e:
  387. log.exception(e)
  388. prompt = user_message
  389. system_message_content = ""
  390. try:
  391. images = await image_generations(
  392. request=request,
  393. form_data=GenerateImageForm(**{"prompt": prompt}),
  394. user=user,
  395. )
  396. await __event_emitter__(
  397. {
  398. "type": "status",
  399. "data": {"description": "Generated an image", "done": True},
  400. }
  401. )
  402. for image in images:
  403. await __event_emitter__(
  404. {
  405. "type": "message",
  406. "data": {"content": f"![Generated Image]({image['url']})\n"},
  407. }
  408. )
  409. system_message_content = "<context>User is shown the generated image, tell the user that the image has been generated</context>"
  410. except Exception as e:
  411. log.exception(e)
  412. await __event_emitter__(
  413. {
  414. "type": "status",
  415. "data": {
  416. "description": f"An error occurred while generating an image",
  417. "done": True,
  418. },
  419. }
  420. )
  421. system_message_content = "<context>Unable to generate an image, tell the user that an error occurred</context>"
  422. if system_message_content:
  423. form_data["messages"] = add_or_update_system_message(
  424. system_message_content, form_data["messages"]
  425. )
  426. return form_data
  427. async def chat_completion_files_handler(
  428. request: Request, body: dict, user: UserModel
  429. ) -> tuple[dict, dict[str, list]]:
  430. sources = []
  431. if files := body.get("metadata", {}).get("files", None):
  432. try:
  433. queries_response = await generate_queries(
  434. request,
  435. {
  436. "model": body["model"],
  437. "messages": body["messages"],
  438. "type": "retrieval",
  439. },
  440. user,
  441. )
  442. queries_response = queries_response["choices"][0]["message"]["content"]
  443. try:
  444. bracket_start = queries_response.find("{")
  445. bracket_end = queries_response.rfind("}") + 1
  446. if bracket_start == -1 or bracket_end == -1:
  447. raise Exception("No JSON object found in the response")
  448. queries_response = queries_response[bracket_start:bracket_end]
  449. queries_response = json.loads(queries_response)
  450. except Exception as e:
  451. queries_response = {"queries": [queries_response]}
  452. queries = queries_response.get("queries", [])
  453. except Exception as e:
  454. queries = []
  455. if len(queries) == 0:
  456. queries = [get_last_user_message(body["messages"])]
  457. try:
  458. # Offload get_sources_from_files to a separate thread
  459. loop = asyncio.get_running_loop()
  460. with ThreadPoolExecutor() as executor:
  461. sources = await loop.run_in_executor(
  462. executor,
  463. lambda: get_sources_from_files(
  464. files=files,
  465. queries=queries,
  466. embedding_function=lambda query: request.app.state.EMBEDDING_FUNCTION(
  467. query, user=user
  468. ),
  469. k=request.app.state.config.TOP_K,
  470. reranking_function=request.app.state.rf,
  471. r=request.app.state.config.RELEVANCE_THRESHOLD,
  472. hybrid_search=request.app.state.config.ENABLE_RAG_HYBRID_SEARCH,
  473. ),
  474. )
  475. except Exception as e:
  476. log.exception(e)
  477. log.debug(f"rag_contexts:sources: {sources}")
  478. return body, {"sources": sources}
  479. def apply_params_to_form_data(form_data, model):
  480. params = form_data.pop("params", {})
  481. if model.get("ollama"):
  482. form_data["options"] = params
  483. if "format" in params:
  484. form_data["format"] = params["format"]
  485. if "keep_alive" in params:
  486. form_data["keep_alive"] = params["keep_alive"]
  487. else:
  488. if "seed" in params:
  489. form_data["seed"] = params["seed"]
  490. if "stop" in params:
  491. form_data["stop"] = params["stop"]
  492. if "temperature" in params:
  493. form_data["temperature"] = params["temperature"]
  494. if "max_tokens" in params:
  495. form_data["max_tokens"] = params["max_tokens"]
  496. if "top_p" in params:
  497. form_data["top_p"] = params["top_p"]
  498. if "frequency_penalty" in params:
  499. form_data["frequency_penalty"] = params["frequency_penalty"]
  500. if "reasoning_effort" in params:
  501. form_data["reasoning_effort"] = params["reasoning_effort"]
  502. return form_data
  503. async def process_chat_payload(request, form_data, metadata, user, model):
  504. form_data = apply_params_to_form_data(form_data, model)
  505. log.debug(f"form_data: {form_data}")
  506. event_emitter = get_event_emitter(metadata)
  507. event_call = get_event_call(metadata)
  508. extra_params = {
  509. "__event_emitter__": event_emitter,
  510. "__event_call__": event_call,
  511. "__user__": {
  512. "id": user.id,
  513. "email": user.email,
  514. "name": user.name,
  515. "role": user.role,
  516. },
  517. "__metadata__": metadata,
  518. "__request__": request,
  519. "__model__": model,
  520. }
  521. # Initialize events to store additional event to be sent to the client
  522. # Initialize contexts and citation
  523. if getattr(request.state, "direct", False) and hasattr(request.state, "model"):
  524. models = {
  525. request.state.model["id"]: request.state.model,
  526. }
  527. else:
  528. models = request.app.state.MODELS
  529. task_model_id = get_task_model_id(
  530. form_data["model"],
  531. request.app.state.config.TASK_MODEL,
  532. request.app.state.config.TASK_MODEL_EXTERNAL,
  533. models,
  534. )
  535. events = []
  536. sources = []
  537. user_message = get_last_user_message(form_data["messages"])
  538. model_knowledge = model.get("info", {}).get("meta", {}).get("knowledge", False)
  539. if model_knowledge:
  540. await event_emitter(
  541. {
  542. "type": "status",
  543. "data": {
  544. "action": "knowledge_search",
  545. "query": user_message,
  546. "done": False,
  547. },
  548. }
  549. )
  550. knowledge_files = []
  551. for item in model_knowledge:
  552. if item.get("collection_name"):
  553. knowledge_files.append(
  554. {
  555. "id": item.get("collection_name"),
  556. "name": item.get("name"),
  557. "legacy": True,
  558. }
  559. )
  560. elif item.get("collection_names"):
  561. knowledge_files.append(
  562. {
  563. "name": item.get("name"),
  564. "type": "collection",
  565. "collection_names": item.get("collection_names"),
  566. "legacy": True,
  567. }
  568. )
  569. else:
  570. knowledge_files.append(item)
  571. files = form_data.get("files", [])
  572. files.extend(knowledge_files)
  573. form_data["files"] = files
  574. variables = form_data.pop("variables", None)
  575. # Process the form_data through the pipeline
  576. try:
  577. form_data = await process_pipeline_inlet_filter(
  578. request, form_data, user, models
  579. )
  580. except Exception as e:
  581. raise e
  582. try:
  583. form_data, flags = await process_filter_functions(
  584. request=request,
  585. filter_ids=get_sorted_filter_ids(model),
  586. filter_type="inlet",
  587. form_data=form_data,
  588. extra_params=extra_params,
  589. )
  590. except Exception as e:
  591. raise Exception(f"Error: {e}")
  592. features = form_data.pop("features", None)
  593. if features:
  594. if "web_search" in features and features["web_search"]:
  595. form_data = await chat_web_search_handler(
  596. request, form_data, extra_params, user
  597. )
  598. if "image_generation" in features and features["image_generation"]:
  599. form_data = await chat_image_generation_handler(
  600. request, form_data, extra_params, user
  601. )
  602. if "code_interpreter" in features and features["code_interpreter"]:
  603. form_data["messages"] = add_or_update_user_message(
  604. (
  605. request.app.state.config.CODE_INTERPRETER_PROMPT_TEMPLATE
  606. if request.app.state.config.CODE_INTERPRETER_PROMPT_TEMPLATE != ""
  607. else DEFAULT_CODE_INTERPRETER_PROMPT
  608. ),
  609. form_data["messages"],
  610. )
  611. tool_ids = form_data.pop("tool_ids", None)
  612. files = form_data.pop("files", None)
  613. # Remove files duplicates
  614. if files:
  615. files = list({json.dumps(f, sort_keys=True): f for f in files}.values())
  616. metadata = {
  617. **metadata,
  618. "tool_ids": tool_ids,
  619. "files": files,
  620. }
  621. form_data["metadata"] = metadata
  622. tool_ids = metadata.get("tool_ids", None)
  623. log.debug(f"{tool_ids=}")
  624. if tool_ids:
  625. # If tool_ids field is present, then get the tools
  626. tools = get_tools(
  627. request,
  628. tool_ids,
  629. user,
  630. {
  631. **extra_params,
  632. "__model__": models[task_model_id],
  633. "__messages__": form_data["messages"],
  634. "__files__": metadata.get("files", []),
  635. },
  636. )
  637. log.info(f"{tools=}")
  638. if metadata.get("function_calling") == "native":
  639. # If the function calling is native, then call the tools function calling handler
  640. metadata["tools"] = tools
  641. form_data["tools"] = [
  642. {"type": "function", "function": tool.get("spec", {})}
  643. for tool in tools.values()
  644. ]
  645. else:
  646. # If the function calling is not native, then call the tools function calling handler
  647. try:
  648. form_data, flags = await chat_completion_tools_handler(
  649. request, form_data, user, models, tools
  650. )
  651. sources.extend(flags.get("sources", []))
  652. except Exception as e:
  653. log.exception(e)
  654. try:
  655. form_data, flags = await chat_completion_files_handler(request, form_data, user)
  656. sources.extend(flags.get("sources", []))
  657. except Exception as e:
  658. log.exception(e)
  659. # If context is not empty, insert it into the messages
  660. if len(sources) > 0:
  661. context_string = ""
  662. for source_idx, source in enumerate(sources):
  663. source_id = source.get("source", {}).get("name", "")
  664. if "document" in source:
  665. for doc_idx, doc_context in enumerate(source["document"]):
  666. context_string += f"<source><source_id>{source_idx}</source_id><source_context>{doc_context}</source_context></source>\n"
  667. context_string = context_string.strip()
  668. prompt = get_last_user_message(form_data["messages"])
  669. if prompt is None:
  670. raise Exception("No user message found")
  671. if (
  672. request.app.state.config.RELEVANCE_THRESHOLD == 0
  673. and context_string.strip() == ""
  674. ):
  675. log.debug(
  676. f"With a 0 relevancy threshold for RAG, the context cannot be empty"
  677. )
  678. # Workaround for Ollama 2.0+ system prompt issue
  679. # TODO: replace with add_or_update_system_message
  680. if model.get("owned_by") == "ollama":
  681. form_data["messages"] = prepend_to_first_user_message_content(
  682. rag_template(
  683. request.app.state.config.RAG_TEMPLATE, context_string, prompt
  684. ),
  685. form_data["messages"],
  686. )
  687. else:
  688. form_data["messages"] = add_or_update_system_message(
  689. rag_template(
  690. request.app.state.config.RAG_TEMPLATE, context_string, prompt
  691. ),
  692. form_data["messages"],
  693. )
  694. # If there are citations, add them to the data_items
  695. sources = [source for source in sources if source.get("source", {}).get("name", "")]
  696. if len(sources) > 0:
  697. events.append({"sources": sources})
  698. if model_knowledge:
  699. await event_emitter(
  700. {
  701. "type": "status",
  702. "data": {
  703. "action": "knowledge_search",
  704. "query": user_message,
  705. "done": True,
  706. "hidden": True,
  707. },
  708. }
  709. )
  710. return form_data, metadata, events
  711. async def process_chat_response(
  712. request, response, form_data, user, events, metadata, tasks
  713. ):
  714. async def background_tasks_handler():
  715. message_map = Chats.get_messages_by_chat_id(metadata["chat_id"])
  716. message = message_map.get(metadata["message_id"]) if message_map else None
  717. if message:
  718. messages = get_message_list(message_map, message.get("id"))
  719. if tasks and messages:
  720. if TASKS.TITLE_GENERATION in tasks:
  721. if tasks[TASKS.TITLE_GENERATION]:
  722. res = await generate_title(
  723. request,
  724. {
  725. "model": message["model"],
  726. "messages": messages,
  727. "chat_id": metadata["chat_id"],
  728. },
  729. user,
  730. )
  731. if res and isinstance(res, dict):
  732. if len(res.get("choices", [])) == 1:
  733. title_string = (
  734. res.get("choices", [])[0]
  735. .get("message", {})
  736. .get("content", message.get("content", "New Chat"))
  737. )
  738. else:
  739. title_string = ""
  740. title_string = title_string[
  741. title_string.find("{") : title_string.rfind("}") + 1
  742. ]
  743. try:
  744. title = json.loads(title_string).get(
  745. "title", "New Chat"
  746. )
  747. except Exception as e:
  748. title = ""
  749. if not title:
  750. title = messages[0].get("content", "New Chat")
  751. Chats.update_chat_title_by_id(metadata["chat_id"], title)
  752. await event_emitter(
  753. {
  754. "type": "chat:title",
  755. "data": title,
  756. }
  757. )
  758. elif len(messages) == 2:
  759. title = messages[0].get("content", "New Chat")
  760. Chats.update_chat_title_by_id(metadata["chat_id"], title)
  761. await event_emitter(
  762. {
  763. "type": "chat:title",
  764. "data": message.get("content", "New Chat"),
  765. }
  766. )
  767. if TASKS.TAGS_GENERATION in tasks and tasks[TASKS.TAGS_GENERATION]:
  768. res = await generate_chat_tags(
  769. request,
  770. {
  771. "model": message["model"],
  772. "messages": messages,
  773. "chat_id": metadata["chat_id"],
  774. },
  775. user,
  776. )
  777. if res and isinstance(res, dict):
  778. if len(res.get("choices", [])) == 1:
  779. tags_string = (
  780. res.get("choices", [])[0]
  781. .get("message", {})
  782. .get("content", "")
  783. )
  784. else:
  785. tags_string = ""
  786. tags_string = tags_string[
  787. tags_string.find("{") : tags_string.rfind("}") + 1
  788. ]
  789. try:
  790. tags = json.loads(tags_string).get("tags", [])
  791. Chats.update_chat_tags_by_id(
  792. metadata["chat_id"], tags, user
  793. )
  794. await event_emitter(
  795. {
  796. "type": "chat:tags",
  797. "data": tags,
  798. }
  799. )
  800. except Exception as e:
  801. pass
  802. event_emitter = None
  803. event_caller = None
  804. if (
  805. "session_id" in metadata
  806. and metadata["session_id"]
  807. and "chat_id" in metadata
  808. and metadata["chat_id"]
  809. and "message_id" in metadata
  810. and metadata["message_id"]
  811. ):
  812. event_emitter = get_event_emitter(metadata)
  813. event_caller = get_event_call(metadata)
  814. # Non-streaming response
  815. if not isinstance(response, StreamingResponse):
  816. if event_emitter:
  817. if "selected_model_id" in response:
  818. Chats.upsert_message_to_chat_by_id_and_message_id(
  819. metadata["chat_id"],
  820. metadata["message_id"],
  821. {
  822. "selectedModelId": response["selected_model_id"],
  823. },
  824. )
  825. if response.get("choices", [])[0].get("message", {}).get("content"):
  826. content = response["choices"][0]["message"]["content"]
  827. if content:
  828. await event_emitter(
  829. {
  830. "type": "chat:completion",
  831. "data": response,
  832. }
  833. )
  834. title = Chats.get_chat_title_by_id(metadata["chat_id"])
  835. await event_emitter(
  836. {
  837. "type": "chat:completion",
  838. "data": {
  839. "done": True,
  840. "content": content,
  841. "title": title,
  842. },
  843. }
  844. )
  845. # Save message in the database
  846. Chats.upsert_message_to_chat_by_id_and_message_id(
  847. metadata["chat_id"],
  848. metadata["message_id"],
  849. {
  850. "content": content,
  851. },
  852. )
  853. # Send a webhook notification if the user is not active
  854. if get_active_status_by_user_id(user.id) is None:
  855. webhook_url = Users.get_user_webhook_url_by_id(user.id)
  856. if webhook_url:
  857. post_webhook(
  858. request.app.state.WEBUI_NAME,
  859. webhook_url,
  860. f"{title} - {request.app.state.config.WEBUI_URL}/c/{metadata['chat_id']}\n\n{content}",
  861. {
  862. "action": "chat",
  863. "message": content,
  864. "title": title,
  865. "url": f"{request.app.state.config.WEBUI_URL}/c/{metadata['chat_id']}",
  866. },
  867. )
  868. await background_tasks_handler()
  869. return response
  870. else:
  871. return response
  872. # Non standard response
  873. if not any(
  874. content_type in response.headers["Content-Type"]
  875. for content_type in ["text/event-stream", "application/x-ndjson"]
  876. ):
  877. return response
  878. # Streaming response
  879. if event_emitter and event_caller:
  880. task_id = str(uuid4()) # Create a unique task ID.
  881. model_id = form_data.get("model", "")
  882. Chats.upsert_message_to_chat_by_id_and_message_id(
  883. metadata["chat_id"],
  884. metadata["message_id"],
  885. {
  886. "model": model_id,
  887. },
  888. )
  889. def split_content_and_whitespace(content):
  890. content_stripped = content.rstrip()
  891. original_whitespace = (
  892. content[len(content_stripped) :]
  893. if len(content) > len(content_stripped)
  894. else ""
  895. )
  896. return content_stripped, original_whitespace
  897. def is_opening_code_block(content):
  898. backtick_segments = content.split("```")
  899. # Even number of segments means the last backticks are opening a new block
  900. return len(backtick_segments) > 1 and len(backtick_segments) % 2 == 0
  901. # Handle as a background task
  902. async def post_response_handler(response, events):
  903. def serialize_content_blocks(content_blocks, raw=False):
  904. content = ""
  905. for block in content_blocks:
  906. if block["type"] == "text":
  907. content = f"{content}{block['content'].strip()}\n"
  908. elif block["type"] == "tool_calls":
  909. attributes = block.get("attributes", {})
  910. block_content = block.get("content", [])
  911. results = block.get("results", [])
  912. if results:
  913. result_display_content = ""
  914. for result in results:
  915. tool_call_id = result.get("tool_call_id", "")
  916. tool_name = ""
  917. for tool_call in block_content:
  918. if tool_call.get("id", "") == tool_call_id:
  919. tool_name = tool_call.get("function", {}).get(
  920. "name", ""
  921. )
  922. break
  923. result_display_content = f"{result_display_content}\n> {tool_name}: {result.get('content', '')}"
  924. if not raw:
  925. content = f'{content}\n<details type="tool_calls" done="true" content="{html.escape(json.dumps(block_content))}" results="{html.escape(json.dumps(results))}">\n<summary>Tool Executed</summary>\n{result_display_content}\n</details>\n'
  926. else:
  927. tool_calls_display_content = ""
  928. for tool_call in block_content:
  929. tool_calls_display_content = f"{tool_calls_display_content}\n> Executing {tool_call.get('function', {}).get('name', '')}"
  930. if not raw:
  931. content = f'{content}\n<details type="tool_calls" done="false" content="{html.escape(json.dumps(block_content))}">\n<summary>Tool Executing...</summary>\n{tool_calls_display_content}\n</details>\n'
  932. elif block["type"] == "reasoning":
  933. reasoning_display_content = "\n".join(
  934. (f"> {line}" if not line.startswith(">") else line)
  935. for line in block["content"].splitlines()
  936. )
  937. reasoning_duration = block.get("duration", None)
  938. if reasoning_duration is not None:
  939. if raw:
  940. content = f'{content}\n<{block["tag"]}>{block["content"]}</{block["tag"]}>\n'
  941. else:
  942. content = f'{content}\n<details type="reasoning" done="true" duration="{reasoning_duration}">\n<summary>Thought for {reasoning_duration} seconds</summary>\n{reasoning_display_content}\n</details>\n'
  943. else:
  944. if raw:
  945. content = f'{content}\n<{block["tag"]}>{block["content"]}</{block["tag"]}>\n'
  946. else:
  947. content = f'{content}\n<details type="reasoning" done="false">\n<summary>Thinking…</summary>\n{reasoning_display_content}\n</details>\n'
  948. elif block["type"] == "code_interpreter":
  949. attributes = block.get("attributes", {})
  950. output = block.get("output", None)
  951. lang = attributes.get("lang", "")
  952. content_stripped, original_whitespace = (
  953. split_content_and_whitespace(content)
  954. )
  955. if is_opening_code_block(content_stripped):
  956. # Remove trailing backticks that would open a new block
  957. content = (
  958. content_stripped.rstrip("`").rstrip()
  959. + original_whitespace
  960. )
  961. else:
  962. # Keep content as is - either closing backticks or no backticks
  963. content = content_stripped + original_whitespace
  964. if output:
  965. output = html.escape(json.dumps(output))
  966. if raw:
  967. content = f'{content}\n<code_interpreter type="code" lang="{lang}">\n{block["content"]}\n</code_interpreter>\n```output\n{output}\n```\n'
  968. else:
  969. content = f'{content}\n<details type="code_interpreter" done="true" output="{output}">\n<summary>Analyzed</summary>\n```{lang}\n{block["content"]}\n```\n</details>\n'
  970. else:
  971. if raw:
  972. content = f'{content}\n<code_interpreter type="code" lang="{lang}">\n{block["content"]}\n</code_interpreter>\n'
  973. else:
  974. content = f'{content}\n<details type="code_interpreter" done="false">\n<summary>Analyzing...</summary>\n```{lang}\n{block["content"]}\n```\n</details>\n'
  975. else:
  976. block_content = str(block["content"]).strip()
  977. content = f"{content}{block['type']}: {block_content}\n"
  978. return content.strip()
  979. def convert_content_blocks_to_messages(content_blocks):
  980. messages = []
  981. temp_blocks = []
  982. for idx, block in enumerate(content_blocks):
  983. if block["type"] == "tool_calls":
  984. messages.append(
  985. {
  986. "role": "assistant",
  987. "content": serialize_content_blocks(temp_blocks),
  988. "tool_calls": block.get("content"),
  989. }
  990. )
  991. results = block.get("results", [])
  992. for result in results:
  993. messages.append(
  994. {
  995. "role": "tool",
  996. "tool_call_id": result["tool_call_id"],
  997. "content": result["content"],
  998. }
  999. )
  1000. temp_blocks = []
  1001. else:
  1002. temp_blocks.append(block)
  1003. if temp_blocks:
  1004. content = serialize_content_blocks(temp_blocks)
  1005. if content:
  1006. messages.append(
  1007. {
  1008. "role": "assistant",
  1009. "content": content,
  1010. }
  1011. )
  1012. return messages
  1013. def tag_content_handler(content_type, tags, content, content_blocks):
  1014. end_flag = False
  1015. def extract_attributes(tag_content):
  1016. """Extract attributes from a tag if they exist."""
  1017. attributes = {}
  1018. if not tag_content: # Ensure tag_content is not None
  1019. return attributes
  1020. # Match attributes in the format: key="value" (ignores single quotes for simplicity)
  1021. matches = re.findall(r'(\w+)\s*=\s*"([^"]+)"', tag_content)
  1022. for key, value in matches:
  1023. attributes[key] = value
  1024. return attributes
  1025. if content_blocks[-1]["type"] == "text":
  1026. for tag in tags:
  1027. # Match start tag e.g., <tag> or <tag attr="value">
  1028. start_tag_pattern = rf"<{tag}(\s.*?)?>"
  1029. match = re.search(start_tag_pattern, content)
  1030. if match:
  1031. attr_content = (
  1032. match.group(1) if match.group(1) else ""
  1033. ) # Ensure it's not None
  1034. attributes = extract_attributes(
  1035. attr_content
  1036. ) # Extract attributes safely
  1037. # Capture everything before and after the matched tag
  1038. before_tag = content[
  1039. : match.start()
  1040. ] # Content before opening tag
  1041. after_tag = content[
  1042. match.end() :
  1043. ] # Content after opening tag
  1044. # Remove the start tag and after from the currently handling text block
  1045. content_blocks[-1]["content"] = content_blocks[-1][
  1046. "content"
  1047. ].replace(match.group(0) + after_tag, "")
  1048. if before_tag:
  1049. content_blocks[-1]["content"] = before_tag
  1050. if not content_blocks[-1]["content"]:
  1051. content_blocks.pop()
  1052. # Append the new block
  1053. content_blocks.append(
  1054. {
  1055. "type": content_type,
  1056. "tag": tag,
  1057. "attributes": attributes,
  1058. "content": "",
  1059. "started_at": time.time(),
  1060. }
  1061. )
  1062. if after_tag:
  1063. content_blocks[-1]["content"] = after_tag
  1064. break
  1065. elif content_blocks[-1]["type"] == content_type:
  1066. tag = content_blocks[-1]["tag"]
  1067. # Match end tag e.g., </tag>
  1068. end_tag_pattern = rf"</{tag}>"
  1069. # Check if the content has the end tag
  1070. if re.search(end_tag_pattern, content):
  1071. end_flag = True
  1072. block_content = content_blocks[-1]["content"]
  1073. # Strip start and end tags from the content
  1074. start_tag_pattern = rf"<{tag}(.*?)>"
  1075. block_content = re.sub(
  1076. start_tag_pattern, "", block_content
  1077. ).strip()
  1078. end_tag_regex = re.compile(end_tag_pattern, re.DOTALL)
  1079. split_content = end_tag_regex.split(block_content, maxsplit=1)
  1080. # Content inside the tag
  1081. block_content = (
  1082. split_content[0].strip() if split_content else ""
  1083. )
  1084. # Leftover content (everything after `</tag>`)
  1085. leftover_content = (
  1086. split_content[1].strip() if len(split_content) > 1 else ""
  1087. )
  1088. if block_content:
  1089. content_blocks[-1]["content"] = block_content
  1090. content_blocks[-1]["ended_at"] = time.time()
  1091. content_blocks[-1]["duration"] = int(
  1092. content_blocks[-1]["ended_at"]
  1093. - content_blocks[-1]["started_at"]
  1094. )
  1095. # Reset the content_blocks by appending a new text block
  1096. if content_type != "code_interpreter":
  1097. if leftover_content:
  1098. content_blocks.append(
  1099. {
  1100. "type": "text",
  1101. "content": leftover_content,
  1102. }
  1103. )
  1104. else:
  1105. content_blocks.append(
  1106. {
  1107. "type": "text",
  1108. "content": "",
  1109. }
  1110. )
  1111. else:
  1112. # Remove the block if content is empty
  1113. content_blocks.pop()
  1114. if leftover_content:
  1115. content_blocks.append(
  1116. {
  1117. "type": "text",
  1118. "content": leftover_content,
  1119. }
  1120. )
  1121. else:
  1122. content_blocks.append(
  1123. {
  1124. "type": "text",
  1125. "content": "",
  1126. }
  1127. )
  1128. # Clean processed content
  1129. content = re.sub(
  1130. rf"<{tag}(.*?)>(.|\n)*?</{tag}>",
  1131. "",
  1132. content,
  1133. flags=re.DOTALL,
  1134. )
  1135. return content, content_blocks, end_flag
  1136. message = Chats.get_message_by_id_and_message_id(
  1137. metadata["chat_id"], metadata["message_id"]
  1138. )
  1139. tool_calls = []
  1140. last_assistant_message = get_last_assistant_message(form_data["messages"])
  1141. content = (
  1142. message.get("content", "")
  1143. if message
  1144. else last_assistant_message if last_assistant_message else ""
  1145. )
  1146. content_blocks = [
  1147. {
  1148. "type": "text",
  1149. "content": content,
  1150. }
  1151. ]
  1152. # We might want to disable this by default
  1153. DETECT_REASONING = True
  1154. DETECT_CODE_INTERPRETER = metadata.get("features", {}).get(
  1155. "code_interpreter", False
  1156. )
  1157. reasoning_tags = [
  1158. "think",
  1159. "thinking",
  1160. "reason",
  1161. "reasoning",
  1162. "thought",
  1163. "Thought",
  1164. ]
  1165. code_interpreter_tags = ["code_interpreter"]
  1166. try:
  1167. for event in events:
  1168. await event_emitter(
  1169. {
  1170. "type": "chat:completion",
  1171. "data": event,
  1172. }
  1173. )
  1174. # Save message in the database
  1175. Chats.upsert_message_to_chat_by_id_and_message_id(
  1176. metadata["chat_id"],
  1177. metadata["message_id"],
  1178. {
  1179. **event,
  1180. },
  1181. )
  1182. async def stream_body_handler(response):
  1183. nonlocal content
  1184. nonlocal content_blocks
  1185. response_tool_calls = []
  1186. async for line in response.body_iterator:
  1187. line = line.decode("utf-8") if isinstance(line, bytes) else line
  1188. data = line
  1189. # Skip empty lines
  1190. if not data.strip():
  1191. continue
  1192. # "data:" is the prefix for each event
  1193. if not data.startswith("data:"):
  1194. continue
  1195. # Remove the prefix
  1196. data = data[len("data:") :].strip()
  1197. try:
  1198. data = json.loads(data)
  1199. if "selected_model_id" in data:
  1200. model_id = data["selected_model_id"]
  1201. Chats.upsert_message_to_chat_by_id_and_message_id(
  1202. metadata["chat_id"],
  1203. metadata["message_id"],
  1204. {
  1205. "selectedModelId": model_id,
  1206. },
  1207. )
  1208. else:
  1209. choices = data.get("choices", [])
  1210. if not choices:
  1211. continue
  1212. delta = choices[0].get("delta", {})
  1213. delta_tool_calls = delta.get("tool_calls", None)
  1214. if delta_tool_calls:
  1215. for delta_tool_call in delta_tool_calls:
  1216. tool_call_index = delta_tool_call.get("index")
  1217. if tool_call_index is not None:
  1218. if (
  1219. len(response_tool_calls)
  1220. <= tool_call_index
  1221. ):
  1222. response_tool_calls.append(
  1223. delta_tool_call
  1224. )
  1225. else:
  1226. delta_name = delta_tool_call.get(
  1227. "function", {}
  1228. ).get("name")
  1229. delta_arguments = delta_tool_call.get(
  1230. "function", {}
  1231. ).get("arguments")
  1232. if delta_name:
  1233. response_tool_calls[
  1234. tool_call_index
  1235. ]["function"]["name"] += delta_name
  1236. if delta_arguments:
  1237. response_tool_calls[
  1238. tool_call_index
  1239. ]["function"][
  1240. "arguments"
  1241. ] += delta_arguments
  1242. value = delta.get("content")
  1243. if value:
  1244. content = f"{content}{value}"
  1245. if not content_blocks:
  1246. content_blocks.append(
  1247. {
  1248. "type": "text",
  1249. "content": "",
  1250. }
  1251. )
  1252. content_blocks[-1]["content"] = (
  1253. content_blocks[-1]["content"] + value
  1254. )
  1255. if DETECT_REASONING:
  1256. content, content_blocks, _ = (
  1257. tag_content_handler(
  1258. "reasoning",
  1259. reasoning_tags,
  1260. content,
  1261. content_blocks,
  1262. )
  1263. )
  1264. if DETECT_CODE_INTERPRETER:
  1265. content, content_blocks, end = (
  1266. tag_content_handler(
  1267. "code_interpreter",
  1268. code_interpreter_tags,
  1269. content,
  1270. content_blocks,
  1271. )
  1272. )
  1273. if end:
  1274. break
  1275. if ENABLE_REALTIME_CHAT_SAVE:
  1276. # Save message in the database
  1277. Chats.upsert_message_to_chat_by_id_and_message_id(
  1278. metadata["chat_id"],
  1279. metadata["message_id"],
  1280. {
  1281. "content": serialize_content_blocks(
  1282. content_blocks
  1283. ),
  1284. },
  1285. )
  1286. else:
  1287. data = {
  1288. "content": serialize_content_blocks(
  1289. content_blocks
  1290. ),
  1291. }
  1292. await event_emitter(
  1293. {
  1294. "type": "chat:completion",
  1295. "data": data,
  1296. }
  1297. )
  1298. except Exception as e:
  1299. done = "data: [DONE]" in line
  1300. if done:
  1301. pass
  1302. else:
  1303. log.debug("Error: ", e)
  1304. continue
  1305. if content_blocks:
  1306. # Clean up the last text block
  1307. if content_blocks[-1]["type"] == "text":
  1308. content_blocks[-1]["content"] = content_blocks[-1][
  1309. "content"
  1310. ].strip()
  1311. if not content_blocks[-1]["content"]:
  1312. content_blocks.pop()
  1313. if not content_blocks:
  1314. content_blocks.append(
  1315. {
  1316. "type": "text",
  1317. "content": "",
  1318. }
  1319. )
  1320. if response_tool_calls:
  1321. tool_calls.append(response_tool_calls)
  1322. if response.background:
  1323. await response.background()
  1324. await stream_body_handler(response)
  1325. MAX_TOOL_CALL_RETRIES = 5
  1326. tool_call_retries = 0
  1327. while len(tool_calls) > 0 and tool_call_retries < MAX_TOOL_CALL_RETRIES:
  1328. tool_call_retries += 1
  1329. response_tool_calls = tool_calls.pop(0)
  1330. content_blocks.append(
  1331. {
  1332. "type": "tool_calls",
  1333. "content": response_tool_calls,
  1334. }
  1335. )
  1336. await event_emitter(
  1337. {
  1338. "type": "chat:completion",
  1339. "data": {
  1340. "content": serialize_content_blocks(content_blocks),
  1341. },
  1342. }
  1343. )
  1344. tools = metadata.get("tools", {})
  1345. results = []
  1346. for tool_call in response_tool_calls:
  1347. tool_call_id = tool_call.get("id", "")
  1348. tool_name = tool_call.get("function", {}).get("name", "")
  1349. tool_function_params = {}
  1350. try:
  1351. # json.loads cannot be used because some models do not produce valid JSON
  1352. tool_function_params = ast.literal_eval(
  1353. tool_call.get("function", {}).get("arguments", "{}")
  1354. )
  1355. except Exception as e:
  1356. log.debug(e)
  1357. tool_result = None
  1358. if tool_name in tools:
  1359. tool = tools[tool_name]
  1360. spec = tool.get("spec", {})
  1361. try:
  1362. required_params = spec.get("parameters", {}).get(
  1363. "required", []
  1364. )
  1365. tool_function = tool["callable"]
  1366. tool_function_params = {
  1367. k: v
  1368. for k, v in tool_function_params.items()
  1369. if k in required_params
  1370. }
  1371. tool_result = await tool_function(
  1372. **tool_function_params
  1373. )
  1374. except Exception as e:
  1375. tool_result = str(e)
  1376. results.append(
  1377. {
  1378. "tool_call_id": tool_call_id,
  1379. "content": tool_result,
  1380. }
  1381. )
  1382. content_blocks[-1]["results"] = results
  1383. content_blocks.append(
  1384. {
  1385. "type": "text",
  1386. "content": "",
  1387. }
  1388. )
  1389. await event_emitter(
  1390. {
  1391. "type": "chat:completion",
  1392. "data": {
  1393. "content": serialize_content_blocks(content_blocks),
  1394. },
  1395. }
  1396. )
  1397. try:
  1398. res = await generate_chat_completion(
  1399. request,
  1400. {
  1401. "model": model_id,
  1402. "stream": True,
  1403. "tools": form_data["tools"],
  1404. "messages": [
  1405. *form_data["messages"],
  1406. *convert_content_blocks_to_messages(content_blocks),
  1407. ],
  1408. },
  1409. user,
  1410. )
  1411. if isinstance(res, StreamingResponse):
  1412. await stream_body_handler(res)
  1413. else:
  1414. break
  1415. except Exception as e:
  1416. log.debug(e)
  1417. break
  1418. if DETECT_CODE_INTERPRETER:
  1419. MAX_RETRIES = 5
  1420. retries = 0
  1421. while (
  1422. content_blocks[-1]["type"] == "code_interpreter"
  1423. and retries < MAX_RETRIES
  1424. ):
  1425. await event_emitter(
  1426. {
  1427. "type": "chat:completion",
  1428. "data": {
  1429. "content": serialize_content_blocks(content_blocks),
  1430. },
  1431. }
  1432. )
  1433. retries += 1
  1434. log.debug(f"Attempt count: {retries}")
  1435. output = ""
  1436. try:
  1437. if content_blocks[-1]["attributes"].get("type") == "code":
  1438. code = content_blocks[-1]["content"]
  1439. if (
  1440. request.app.state.config.CODE_INTERPRETER_ENGINE
  1441. == "pyodide"
  1442. ):
  1443. output = await event_caller(
  1444. {
  1445. "type": "execute:python",
  1446. "data": {
  1447. "id": str(uuid4()),
  1448. "code": code,
  1449. "session_id": metadata.get(
  1450. "session_id", None
  1451. ),
  1452. },
  1453. }
  1454. )
  1455. elif (
  1456. request.app.state.config.CODE_INTERPRETER_ENGINE
  1457. == "jupyter"
  1458. ):
  1459. output = await execute_code_jupyter(
  1460. request.app.state.config.CODE_INTERPRETER_JUPYTER_URL,
  1461. code,
  1462. (
  1463. request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH_TOKEN
  1464. if request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH
  1465. == "token"
  1466. else None
  1467. ),
  1468. (
  1469. request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH_PASSWORD
  1470. if request.app.state.config.CODE_INTERPRETER_JUPYTER_AUTH
  1471. == "password"
  1472. else None
  1473. ),
  1474. )
  1475. else:
  1476. output = {
  1477. "stdout": "Code interpreter engine not configured."
  1478. }
  1479. log.debug(f"Code interpreter output: {output}")
  1480. if isinstance(output, dict):
  1481. stdout = output.get("stdout", "")
  1482. if isinstance(stdout, str):
  1483. stdoutLines = stdout.split("\n")
  1484. for idx, line in enumerate(stdoutLines):
  1485. if "data:image/png;base64" in line:
  1486. id = str(uuid4())
  1487. # ensure the path exists
  1488. os.makedirs(
  1489. os.path.join(CACHE_DIR, "images"),
  1490. exist_ok=True,
  1491. )
  1492. image_path = os.path.join(
  1493. CACHE_DIR,
  1494. f"images/{id}.png",
  1495. )
  1496. with open(image_path, "wb") as f:
  1497. f.write(
  1498. base64.b64decode(
  1499. line.split(",")[1]
  1500. )
  1501. )
  1502. stdoutLines[idx] = (
  1503. f"![Output Image {idx}](/cache/images/{id}.png)"
  1504. )
  1505. output["stdout"] = "\n".join(stdoutLines)
  1506. result = output.get("result", "")
  1507. if isinstance(result, str):
  1508. resultLines = result.split("\n")
  1509. for idx, line in enumerate(resultLines):
  1510. if "data:image/png;base64" in line:
  1511. id = str(uuid4())
  1512. # ensure the path exists
  1513. os.makedirs(
  1514. os.path.join(CACHE_DIR, "images"),
  1515. exist_ok=True,
  1516. )
  1517. image_path = os.path.join(
  1518. CACHE_DIR,
  1519. f"images/{id}.png",
  1520. )
  1521. with open(image_path, "wb") as f:
  1522. f.write(
  1523. base64.b64decode(
  1524. line.split(",")[1]
  1525. )
  1526. )
  1527. resultLines[idx] = (
  1528. f"![Output Image {idx}](/cache/images/{id}.png)"
  1529. )
  1530. output["result"] = "\n".join(resultLines)
  1531. except Exception as e:
  1532. output = str(e)
  1533. content_blocks[-1]["output"] = output
  1534. content_blocks.append(
  1535. {
  1536. "type": "text",
  1537. "content": "",
  1538. }
  1539. )
  1540. await event_emitter(
  1541. {
  1542. "type": "chat:completion",
  1543. "data": {
  1544. "content": serialize_content_blocks(content_blocks),
  1545. },
  1546. }
  1547. )
  1548. print(content_blocks, serialize_content_blocks(content_blocks))
  1549. try:
  1550. res = await generate_chat_completion(
  1551. request,
  1552. {
  1553. "model": model_id,
  1554. "stream": True,
  1555. "messages": [
  1556. *form_data["messages"],
  1557. {
  1558. "role": "assistant",
  1559. "content": serialize_content_blocks(
  1560. content_blocks, raw=True
  1561. ),
  1562. },
  1563. ],
  1564. },
  1565. user,
  1566. )
  1567. if isinstance(res, StreamingResponse):
  1568. await stream_body_handler(res)
  1569. else:
  1570. break
  1571. except Exception as e:
  1572. log.debug(e)
  1573. break
  1574. title = Chats.get_chat_title_by_id(metadata["chat_id"])
  1575. data = {
  1576. "done": True,
  1577. "content": serialize_content_blocks(content_blocks),
  1578. "title": title,
  1579. }
  1580. if not ENABLE_REALTIME_CHAT_SAVE:
  1581. # Save message in the database
  1582. Chats.upsert_message_to_chat_by_id_and_message_id(
  1583. metadata["chat_id"],
  1584. metadata["message_id"],
  1585. {
  1586. "content": serialize_content_blocks(content_blocks),
  1587. },
  1588. )
  1589. # Send a webhook notification if the user is not active
  1590. if get_active_status_by_user_id(user.id) is None:
  1591. webhook_url = Users.get_user_webhook_url_by_id(user.id)
  1592. if webhook_url:
  1593. post_webhook(
  1594. request.app.state.WEBUI_NAME,
  1595. webhook_url,
  1596. f"{title} - {request.app.state.config.WEBUI_URL}/c/{metadata['chat_id']}\n\n{content}",
  1597. {
  1598. "action": "chat",
  1599. "message": content,
  1600. "title": title,
  1601. "url": f"{request.app.state.config.WEBUI_URL}/c/{metadata['chat_id']}",
  1602. },
  1603. )
  1604. await event_emitter(
  1605. {
  1606. "type": "chat:completion",
  1607. "data": data,
  1608. }
  1609. )
  1610. await background_tasks_handler()
  1611. except asyncio.CancelledError:
  1612. print("Task was cancelled!")
  1613. await event_emitter({"type": "task-cancelled"})
  1614. if not ENABLE_REALTIME_CHAT_SAVE:
  1615. # Save message in the database
  1616. Chats.upsert_message_to_chat_by_id_and_message_id(
  1617. metadata["chat_id"],
  1618. metadata["message_id"],
  1619. {
  1620. "content": serialize_content_blocks(content_blocks),
  1621. },
  1622. )
  1623. if response.background is not None:
  1624. await response.background()
  1625. # background_tasks.add_task(post_response_handler, response, events)
  1626. task_id, _ = create_task(post_response_handler(response, events))
  1627. return {"status": True, "task_id": task_id}
  1628. else:
  1629. # Fallback to the original response
  1630. async def stream_wrapper(original_generator, events):
  1631. def wrap_item(item):
  1632. return f"data: {item}\n\n"
  1633. for event in events:
  1634. yield wrap_item(json.dumps(event))
  1635. async for data in original_generator:
  1636. yield data
  1637. return StreamingResponse(
  1638. stream_wrapper(response.body_iterator, events),
  1639. headers=dict(response.headers),
  1640. background=response.background,
  1641. )