steps.py 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799
  1. import asyncio
  2. import collections
  3. import json
  4. import os
  5. import re
  6. import socket
  7. import subprocess
  8. import time
  9. from contextlib import closing
  10. from re import RegexFlag
  11. import aiohttp
  12. import openai
  13. from behave import step
  14. from behave.api.async_step import async_run_until_complete
  15. from prometheus_client import parser
  16. @step(u"a server listening on {server_fqdn}:{server_port}")
  17. def step_server_config(context, server_fqdn, server_port):
  18. context.server_fqdn = server_fqdn
  19. context.server_port = int(server_port)
  20. if 'PORT' in os.environ:
  21. context.server_port = int(os.environ['PORT'])
  22. print(f"$PORT set, overriding server port with to {context.server_port}")
  23. context.base_url = f'http://{context.server_fqdn}:{context.server_port}'
  24. context.debug = 'DEBUG' in os.environ and os.environ['DEBUG'] == 'ON'
  25. context.model_alias = None
  26. context.n_ctx = None
  27. context.n_predict = None
  28. context.n_server_predict = None
  29. context.n_slots = None
  30. context.server_api_key = None
  31. context.server_continuous_batching = False
  32. context.server_embeddings = False
  33. context.server_metrics = False
  34. context.server_process = None
  35. context.server_seed = None
  36. context.user_api_key = None
  37. context.tasks_result = []
  38. context.concurrent_tasks = []
  39. context.prompts = []
  40. @step(u'a model file {model_file}')
  41. def step_model_file(context, model_file):
  42. context.model_file = model_file
  43. @step(u'a model alias {model_alias}')
  44. def step_model_alias(context, model_alias):
  45. context.model_alias = model_alias
  46. @step(u'{seed} as server seed')
  47. def step_seed(context, seed):
  48. context.server_seed = int(seed)
  49. @step(u'{n_ctx} KV cache size')
  50. def step_n_ctx(context, n_ctx):
  51. context.n_ctx = int(n_ctx)
  52. @step(u'{n_slots} slots')
  53. def step_n_slots(context, n_slots):
  54. context.n_slots = int(n_slots)
  55. @step(u'{n_predict} server max tokens to predict')
  56. def step_server_n_predict(context, n_predict):
  57. context.n_server_predict = int(n_predict)
  58. @step(u'continuous batching')
  59. def step_server_continuous_batching(context):
  60. context.server_continuous_batching = True
  61. @step(u'embeddings extraction')
  62. def step_server_embeddings(context):
  63. context.server_embeddings = True
  64. @step(u'prometheus compatible metrics exposed')
  65. def step_server_metrics(context):
  66. context.server_metrics = True
  67. @step(u"the server is starting")
  68. def step_start_server(context):
  69. start_server_background(context)
  70. attempts = 0
  71. while True:
  72. with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
  73. result = sock.connect_ex((context.server_fqdn, context.server_port))
  74. if result == 0:
  75. print("\x1b[33;46mserver started!\x1b[0m")
  76. return
  77. attempts += 1
  78. if attempts > 20:
  79. assert False, "server not started"
  80. print(f"waiting for server to start, connect error code = {result}...")
  81. time.sleep(0.1)
  82. @step(u"the server is {expecting_status}")
  83. @async_run_until_complete
  84. async def step_wait_for_the_server_to_be_started(context, expecting_status):
  85. match expecting_status:
  86. case 'healthy':
  87. await wait_for_health_status(context, context.base_url, 200, 'ok')
  88. case 'ready' | 'idle':
  89. await wait_for_health_status(context, context.base_url, 200, 'ok',
  90. params={'fail_on_no_slot': 0, 'include_slots': 0},
  91. slots_idle=context.n_slots,
  92. slots_processing=0,
  93. expected_slots=[{'id': slot_id, 'state': 0}
  94. for slot_id in range(context.n_slots)])
  95. case 'busy':
  96. await wait_for_health_status(context, context.base_url, 503,
  97. 'no slot available',
  98. params={'fail_on_no_slot': 0, 'include_slots': 0},
  99. slots_idle=0,
  100. slots_processing=context.n_slots,
  101. expected_slots=[{'id': slot_id, 'state': 1}
  102. for slot_id in range(context.n_slots)])
  103. case _:
  104. assert False, "unknown status"
  105. @step(u'all slots are {expected_slot_status_string}')
  106. @async_run_until_complete
  107. async def step_all_slots_status(context, expected_slot_status_string):
  108. match expected_slot_status_string:
  109. case 'idle':
  110. expected_slot_status = 0
  111. case 'busy':
  112. expected_slot_status = 1
  113. case _:
  114. assert False, "unknown status"
  115. expected_slots = [{'id': slot_id, 'state': expected_slot_status}
  116. for slot_id in range(context.n_slots)]
  117. await request_slots_status(context, expected_slots)
  118. @step(u'a completion request with {api_error} api error')
  119. @async_run_until_complete
  120. async def step_request_completion(context, api_error):
  121. expect_api_error = api_error == 'raised'
  122. completion = await request_completion(context.prompts.pop(),
  123. context.base_url,
  124. debug=context.debug,
  125. n_predict=context.n_predict,
  126. server_seed=context.server_seed,
  127. expect_api_error=expect_api_error,
  128. user_api_key=context.user_api_key)
  129. context.tasks_result.append(completion)
  130. if context.debug:
  131. print(f"Completion response: {completion}")
  132. if expect_api_error:
  133. assert completion == 401, f"completion must be an 401 status code: {completion}"
  134. @step(u'{predicted_n} tokens are predicted matching {re_content}')
  135. def step_n_tokens_predicted_with_content(context, predicted_n, re_content):
  136. assert_n_tokens_predicted(context.tasks_result.pop(), int(predicted_n), re_content)
  137. @step(u'{predicted_n} tokens are predicted')
  138. def step_n_tokens_predicted(context, predicted_n):
  139. assert_n_tokens_predicted(context.tasks_result.pop(), int(predicted_n))
  140. @step(u'a user prompt {user_prompt}')
  141. def step_user_prompt(context, user_prompt):
  142. context.prompts.append(user_prompt)
  143. @step(u'a system prompt {system_prompt}')
  144. def step_system_prompt(context, system_prompt):
  145. context.system_prompt = system_prompt
  146. @step(u'a model {model}')
  147. def step_model(context, model):
  148. context.model = model
  149. @step(u'{max_tokens} max tokens to predict')
  150. def step_max_tokens(context, max_tokens):
  151. context.n_predict = int(max_tokens)
  152. @step(u'streaming is {enable_streaming}')
  153. def step_streaming(context, enable_streaming):
  154. context.enable_streaming = enable_streaming == 'enabled'
  155. @step(u'a user api key {user_api_key}')
  156. def step_user_api_key(context, user_api_key):
  157. context.user_api_key = user_api_key
  158. @step(u'no user api key')
  159. def step_no_user_api_key(context):
  160. context.user_api_key = None
  161. @step(u'a user api key ')
  162. def step_no_user_api_key_space(context):
  163. context.user_api_key = None
  164. @step(u'a server api key {server_api_key}')
  165. def step_server_api_key(context, server_api_key):
  166. context.server_api_key = server_api_key
  167. @step(u'an OAI compatible chat completions request with {api_error} api error')
  168. @async_run_until_complete
  169. async def step_oai_chat_completions(context, api_error):
  170. if context.debug:
  171. print(f"Submitting OAI compatible completions request...")
  172. expect_api_error = api_error == 'raised'
  173. completion = await oai_chat_completions(context.prompts.pop(),
  174. context.system_prompt,
  175. context.base_url,
  176. False,
  177. model=context.model if hasattr(context, 'model') else None,
  178. n_predict=context.n_predict
  179. if hasattr(context, 'n_predict') else None,
  180. enable_streaming=context.enable_streaming
  181. if hasattr(context, 'enable_streaming') else None,
  182. server_seed=context.server_seed
  183. if hasattr(context, 'server_seed') else None,
  184. user_api_key=context.user_api_key
  185. if hasattr(context, 'user_api_key') else None,
  186. expect_api_error=expect_api_error)
  187. context.tasks_result.append(completion)
  188. if context.debug:
  189. print(f"Completion response: {completion}")
  190. if expect_api_error:
  191. assert completion == 401, f"completion must be an 401 status code: {completion}"
  192. if context.debug:
  193. print(f"Completion response: {completion}")
  194. @step(u'a prompt')
  195. def step_a_prompt(context):
  196. context.prompts.append(context.text)
  197. @step(u'a prompt {prompt}')
  198. def step_a_prompt_prompt(context, prompt):
  199. context.prompts.append(prompt)
  200. @step(u'concurrent completion requests')
  201. @async_run_until_complete()
  202. async def step_concurrent_completion_requests(context):
  203. await concurrent_requests(context,
  204. request_completion,
  205. # prompt is inserted automatically
  206. context.base_url,
  207. debug=context.debug,
  208. n_predict=context.n_predict if hasattr(context, 'n_predict') else None,
  209. server_seed=context.server_seed if hasattr(context, 'server_seed') else None,
  210. user_api_key=context.user_api_key if hasattr(context,
  211. 'user_api_key') else None)
  212. @step(u'concurrent OAI completions requests')
  213. @async_run_until_complete
  214. async def step_oai_chat_completions(context):
  215. await concurrent_requests(context, oai_chat_completions,
  216. # user_prompt is inserted automatically
  217. context.system_prompt,
  218. context.base_url,
  219. True, # async_client
  220. model=context.model
  221. if hasattr(context, 'model') else None,
  222. n_predict=context.n_predict
  223. if hasattr(context, 'n_predict') else None,
  224. enable_streaming=context.enable_streaming
  225. if hasattr(context, 'enable_streaming') else None,
  226. server_seed=context.server_seed
  227. if hasattr(context, 'server_seed') else None,
  228. user_api_key=context.user_api_key
  229. if hasattr(context, 'user_api_key') else None)
  230. @step(u'all prompts are predicted')
  231. @async_run_until_complete
  232. async def step_all_prompts_are_predicted(context):
  233. await all_prompts_are_predicted(context)
  234. @step(u'all prompts are predicted with {n_predict} tokens')
  235. @async_run_until_complete
  236. async def step_all_prompts_are_predicted_with_n_tokens(context, n_predict):
  237. expected_predicted_n = int(n_predict)
  238. await all_prompts_are_predicted(context, expected_predicted_n)
  239. async def all_prompts_are_predicted(context, expected_predicted_n=None):
  240. n_completions = await gather_tasks_results(context)
  241. assert n_completions > 0
  242. for i in range(n_completions):
  243. assert_n_tokens_predicted(context.tasks_result.pop(), expected_predicted_n=expected_predicted_n)
  244. assert len(context.concurrent_tasks) == 0, f"{len(context.concurrent_tasks)} pending requests"
  245. @step(u'embeddings are computed for')
  246. @async_run_until_complete
  247. async def step_compute_embedding(context):
  248. context.embeddings = await request_embedding(context.text, base_url=context.base_url)
  249. @step(u'embeddings are generated')
  250. def step_assert_embeddings(context):
  251. if len(context.prompts) == 0:
  252. assert_embeddings(context.embeddings)
  253. else:
  254. assert len(context.embeddings) == len(context.prompts), (f"unexpected response:\n"
  255. f"context.prompts={context.prompts}\n"
  256. f"context.embeddings={context.embeddings}")
  257. for embedding in context.embeddings:
  258. context.prompts.pop()
  259. assert_embeddings(embedding)
  260. @step(u'an OAI compatible embeddings computation request for')
  261. @async_run_until_complete
  262. async def step_oai_compute_embeddings(context):
  263. context.embeddings = await request_oai_embeddings(context.text,
  264. base_url=context.base_url,
  265. user_api_key=context.user_api_key,
  266. model=context.model)
  267. @step(u'an OAI compatible embeddings computation request for multiple inputs')
  268. @async_run_until_complete
  269. async def step_oai_compute_embeddings_multiple_inputs(context):
  270. context.embeddings = await request_oai_embeddings(context.prompts,
  271. base_url=context.base_url,
  272. user_api_key=context.user_api_key,
  273. model=context.model)
  274. @step(u'concurrent embedding requests')
  275. @async_run_until_complete()
  276. async def step_concurrent_embedding_requests(context):
  277. await concurrent_requests(context,
  278. request_embedding,
  279. # prompt is inserted automatically
  280. base_url=context.base_url)
  281. @step(u'concurrent OAI embedding requests')
  282. @async_run_until_complete()
  283. async def step_concurrent_oai_embedding_requests(context):
  284. await concurrent_requests(context,
  285. request_oai_embeddings,
  286. # prompt is inserted automatically
  287. base_url=context.base_url,
  288. async_client=True,
  289. model=context.model)
  290. @step(u'all embeddings are generated')
  291. @async_run_until_complete()
  292. async def all_embeddings_are_generated(context):
  293. n_embedding_requests = await gather_tasks_results(context)
  294. assert n_embedding_requests > 0
  295. for i in range(n_embedding_requests):
  296. assert_embeddings(context.tasks_result.pop())
  297. @step(u'tokenizing')
  298. @async_run_until_complete
  299. async def step_tokenize(context):
  300. context.tokenized_text = context.text
  301. async with aiohttp.ClientSession() as session:
  302. async with session.post(f'{context.base_url}/tokenize',
  303. json={
  304. "content": context.tokenized_text,
  305. }) as response:
  306. assert response.status == 200
  307. tokenize_json = await response.json()
  308. context.tokens = tokenize_json['tokens']
  309. @step(u'tokens can be detokenize')
  310. @async_run_until_complete
  311. async def step_detokenize(context):
  312. assert len(context.tokens) > 0
  313. async with aiohttp.ClientSession() as session:
  314. async with session.post(f'{context.base_url}/detokenize',
  315. json={
  316. "tokens": context.tokens,
  317. }) as response:
  318. assert response.status == 200
  319. detokenize_json = await response.json()
  320. # SPM tokenizer adds a whitespace prefix: https://github.com/google/sentencepiece/issues/15
  321. assert context.tokenized_text == detokenize_json['content'].strip()
  322. @step(u'an OPTIONS request is sent from {origin}')
  323. @async_run_until_complete
  324. async def step_options_request(context, origin):
  325. async with aiohttp.ClientSession() as session:
  326. async with session.options(f'{context.base_url}/v1/chat/completions',
  327. headers={"Origin": origin}) as response:
  328. assert response.status == 200
  329. context.options_response = response
  330. @step(u'CORS header {cors_header} is set to {cors_header_value}')
  331. def step_check_options_header_value(context, cors_header, cors_header_value):
  332. assert context.options_response.headers[cors_header] == cors_header_value
  333. @step(u'prometheus metrics are exposed')
  334. @async_run_until_complete
  335. async def step_prometheus_metrics_exported(context):
  336. async with aiohttp.ClientSession() as session:
  337. async with await session.get(f'{context.base_url}/metrics') as metrics_response:
  338. assert metrics_response.status == 200
  339. assert metrics_response.headers['Content-Type'] == "text/plain; version=0.0.4"
  340. metrics_raw = await metrics_response.text()
  341. metric_exported = False
  342. for metric in parser.text_string_to_metric_families(metrics_raw):
  343. match metric.name:
  344. case "llamacpp:kv_cache_usage_ratio":
  345. assert len(metric.samples) > 0
  346. metric_exported = True
  347. assert metric_exported, "No metrics exported"
  348. async def concurrent_requests(context, f_completion, *args, **kwargs):
  349. n_prompts = len(context.prompts)
  350. if context.debug:
  351. print(f"starting {n_prompts} concurrent completion requests...")
  352. assert n_prompts > 0
  353. for prompt_no in range(n_prompts):
  354. shifted_args = [context.prompts.pop(), *args]
  355. context.concurrent_tasks.append(asyncio.create_task(f_completion(*shifted_args, **kwargs)))
  356. await asyncio.sleep(0.1)
  357. async def request_completion(prompt,
  358. base_url,
  359. debug=False,
  360. n_predict=None,
  361. server_seed=None,
  362. expect_api_error=None,
  363. user_api_key=None):
  364. if debug:
  365. print(f"Sending completion request: {prompt}")
  366. origin = "my.super.domain"
  367. headers = {
  368. 'Origin': origin
  369. }
  370. if user_api_key is not None:
  371. if debug:
  372. print(f"Set user_api_key: {user_api_key}")
  373. headers['Authorization'] = f'Bearer {user_api_key}'
  374. async with aiohttp.ClientSession() as session:
  375. async with session.post(f'{base_url}/completion',
  376. json={
  377. "prompt": prompt,
  378. "n_predict": int(n_predict) if n_predict is not None else -1,
  379. "seed": server_seed if server_seed is not None else 42
  380. },
  381. headers=headers) as response:
  382. if expect_api_error is None or not expect_api_error:
  383. assert response.status == 200
  384. assert response.headers['Access-Control-Allow-Origin'] == origin
  385. return await response.json()
  386. else:
  387. return response.status
  388. async def oai_chat_completions(user_prompt,
  389. system_prompt,
  390. base_url,
  391. async_client,
  392. debug=False,
  393. model=None,
  394. n_predict=None,
  395. enable_streaming=None,
  396. server_seed=None,
  397. user_api_key=None,
  398. expect_api_error=None):
  399. if debug:
  400. print(f"Sending OAI Chat completions request: {user_prompt}")
  401. # openai client always expects an api key
  402. user_api_key = user_api_key if user_api_key is not None else 'nope'
  403. seed = server_seed if server_seed is not None else 42
  404. enable_streaming = enable_streaming if enable_streaming is not None else False
  405. payload = {
  406. "messages": [
  407. {
  408. "role": "system",
  409. "content": system_prompt,
  410. },
  411. {
  412. "role": "user",
  413. "content": user_prompt,
  414. }
  415. ],
  416. "model": model,
  417. "max_tokens": n_predict,
  418. "stream": enable_streaming,
  419. "seed": seed
  420. }
  421. completion_response = {
  422. 'content': '',
  423. 'timings': {
  424. 'predicted_n': 0
  425. }
  426. }
  427. if async_client:
  428. origin = 'llama.cpp'
  429. headers = {'Authorization': f'Bearer {user_api_key}', 'Origin': origin}
  430. async with aiohttp.ClientSession() as session:
  431. async with session.post(f'{base_url}/v1/chat/completions',
  432. json=payload,
  433. headers=headers) as response:
  434. if enable_streaming:
  435. assert response.status == 200
  436. assert response.headers['Access-Control-Allow-Origin'] == origin
  437. assert response.headers['Content-Type'] == "text/event-stream"
  438. event_received = True
  439. while event_received:
  440. event_received = False
  441. async for line_in_bytes in response.content:
  442. line = line_in_bytes.decode('utf8')
  443. line = line.rstrip('\n').rstrip('\r')
  444. if line == '':
  445. continue
  446. event_data = line.split(': ', 1)
  447. assert event_data[0] == 'data', f'Bad event code received: ```{event_data}```'
  448. chunk_raw = event_data[1]
  449. chunk = json.loads(chunk_raw)
  450. assert len(chunk['choices']) == 1, f"no choices provided, line ```{line}```"
  451. delta = chunk['choices'][0]['delta']
  452. if 'content' in delta:
  453. completion_response['content'] += delta['content']
  454. completion_response['timings']['predicted_n'] += 1
  455. else:
  456. if expect_api_error is None or not expect_api_error:
  457. assert response.status == 200
  458. assert response.headers['Access-Control-Allow-Origin'] == origin
  459. assert response.headers['Content-Type'] == "application/json; charset=utf-8"
  460. chat_completion_raw = await response.json()
  461. completion_response = {
  462. 'content': chat_completion_raw['choices'][0]['message'],
  463. 'timings': {
  464. 'predicted_n': chat_completion_raw['usage']['completion_tokens']
  465. }
  466. }
  467. else:
  468. return response.status
  469. else:
  470. try:
  471. openai.api_key = user_api_key
  472. openai.api_base = f'{base_url}/v1/chat'
  473. chat_completion = openai.Completion.create(
  474. messages=payload['messages'],
  475. model=model,
  476. max_tokens=n_predict,
  477. stream=enable_streaming,
  478. seed=seed
  479. )
  480. except openai.error.APIError as e:
  481. if expect_api_error is not None and expect_api_error:
  482. return 401
  483. else:
  484. assert False, f'error raised: {e}'
  485. if enable_streaming:
  486. for chunk in chat_completion:
  487. assert len(chunk.choices) == 1
  488. delta = chunk.choices[0].delta
  489. if 'content' in delta:
  490. completion_response['content'] += delta['content']
  491. completion_response['timings']['predicted_n'] += 1
  492. else:
  493. assert len(chat_completion.choices) == 1
  494. completion_response = {
  495. 'content': chat_completion.choices[0].message.content,
  496. 'timings': {
  497. 'predicted_n': chat_completion.usage.completion_tokens
  498. }
  499. }
  500. if debug:
  501. print("OAI response formatted to llama.cpp:", completion_response)
  502. return completion_response
  503. async def request_embedding(content, base_url=None):
  504. async with aiohttp.ClientSession() as session:
  505. async with session.post(f'{base_url}/embedding',
  506. json={
  507. "content": content,
  508. }) as response:
  509. assert response.status == 200
  510. response_json = await response.json()
  511. return response_json['embedding']
  512. async def request_oai_embeddings(input,
  513. base_url=None, user_api_key=None,
  514. model=None, async_client=False):
  515. # openai client always expects an api_key
  516. user_api_key = user_api_key if user_api_key is not None else 'nope'
  517. if async_client:
  518. origin = 'llama.cpp'
  519. if user_api_key is not None:
  520. headers = {'Authorization': f'Bearer {user_api_key}', 'Origin': origin}
  521. async with aiohttp.ClientSession() as session:
  522. async with session.post(f'{base_url}/v1/embeddings',
  523. json={
  524. "input": input,
  525. "model": model,
  526. },
  527. headers=headers) as response:
  528. assert response.status == 200, f"received status code not expected: {response.status}"
  529. assert response.headers['Access-Control-Allow-Origin'] == origin
  530. assert response.headers['Content-Type'] == "application/json; charset=utf-8"
  531. response_json = await response.json()
  532. assert response_json['model'] == model, f"invalid model received: {response_json['model']}"
  533. assert response_json['object'] == 'list'
  534. return response_json['data']
  535. else:
  536. openai.api_key = user_api_key
  537. openai.api_base = f'{base_url}/v1'
  538. oai_embeddings = openai.Embedding.create(
  539. model=model,
  540. input=input,
  541. )
  542. if isinstance(input, collections.abc.Sequence):
  543. embeddings = []
  544. for an_oai_embeddings in oai_embeddings.data:
  545. embeddings.append(an_oai_embeddings.embedding)
  546. else:
  547. embeddings = oai_embeddings.data.embedding
  548. return embeddings
  549. def assert_n_tokens_predicted(completion_response, expected_predicted_n=None, re_content=None):
  550. content = completion_response['content']
  551. n_predicted = completion_response['timings']['predicted_n']
  552. assert len(content) > 0, "no token predicted"
  553. if expected_predicted_n is not None:
  554. assert n_predicted == expected_predicted_n, (f'invalid number of tokens predicted:'
  555. f' {n_predicted} <> {expected_predicted_n}')
  556. if re_content is not None:
  557. re_content = '^.*' + re_content.replace('<or>', '|') + '.*$'
  558. assert re.match(re_content, content, flags=RegexFlag.IGNORECASE | RegexFlag.MULTILINE | RegexFlag.DOTALL), (
  559. f'invalid tokens predicted:'
  560. f' ```\n{content}\n``` do not match /{re_content}/')
  561. async def gather_tasks_results(context):
  562. n_tasks = len(context.concurrent_tasks)
  563. if context.debug:
  564. print(f"Waiting for all {n_tasks} tasks results...")
  565. for task_no in range(n_tasks):
  566. context.tasks_result.append(await context.concurrent_tasks.pop())
  567. n_completions = len(context.tasks_result)
  568. return n_completions
  569. async def wait_for_health_status(context,
  570. base_url,
  571. expected_http_status_code,
  572. expected_health_status,
  573. params=None,
  574. slots_idle=None,
  575. slots_processing=None,
  576. expected_slots=None):
  577. if context.debug:
  578. print(f"Starting checking for health for expected_health_status={expected_health_status}")
  579. timeout = 3 # seconds
  580. interval = 0.5
  581. counter = 0
  582. async with aiohttp.ClientSession() as session:
  583. while True:
  584. async with await session.get(f'{base_url}/health', params=params) as health_response:
  585. status_code = health_response.status
  586. health = await health_response.json()
  587. if context.debug:
  588. print(f"HEALTH - response for expected health status='{expected_health_status}' on "
  589. f"'{base_url}/health'?{params} is {health}")
  590. if (status_code == expected_http_status_code
  591. and health['status'] == expected_health_status
  592. and (slots_idle is None or health['slots_idle'] == slots_idle)
  593. and (slots_processing is None or health['slots_processing'] == slots_processing)):
  594. if expected_slots is not None:
  595. assert_slots_status(health['slots'], expected_slots)
  596. return
  597. if (status_code == expected_http_status_code
  598. and health['status'] == expected_health_status
  599. and (slots_idle is None or health['slots_idle'] == slots_idle)
  600. and (slots_processing is None or health['slots_processing'] == slots_processing)):
  601. if expected_slots is not None:
  602. assert_slots_status(health['slots'], expected_slots)
  603. return
  604. await asyncio.sleep(interval)
  605. counter += interval
  606. if counter >= timeout:
  607. # Sometimes health requests are triggered after completions are predicted
  608. if expected_http_status_code == 503:
  609. if len(context.tasks_result) == 0:
  610. print("\x1b[5;37;43mWARNING: forcing concurrent tasks,"
  611. " busy health check missed, probably too fast inference\x1b[0m")
  612. n_completions = await gather_tasks_results(context)
  613. if n_completions > 0:
  614. return
  615. assert False, 'timeout exceeded'
  616. def assert_embeddings(embeddings):
  617. assert len(embeddings) > 0
  618. embeddings_computed = False
  619. for emb in embeddings:
  620. if emb != 0:
  621. embeddings_computed = True
  622. assert embeddings_computed, f"Embeddings: {embeddings}"
  623. async def request_slots_status(context, expected_slots):
  624. async with aiohttp.ClientSession() as session:
  625. async with await session.get(f'{context.base_url}/slots') as slots_response:
  626. assert slots_response.status == 200
  627. slots = await slots_response.json()
  628. assert_slots_status(slots, expected_slots)
  629. def assert_slots_status(slots, expected_slots):
  630. assert len(slots) == len(expected_slots)
  631. for slot_id, (expected, slot) in enumerate(zip(expected_slots, slots)):
  632. for key in expected:
  633. assert expected[key] == slot[key], (f"invalid slot {slot_id}"
  634. f" expected[{key}] != slot[{key}]"
  635. f" = {expected[key]} != {slot[key]}")
  636. def start_server_background(context):
  637. context.server_path = '../../../build/bin/server'
  638. if 'LLAMA_SERVER_BIN_PATH' in os.environ:
  639. context.server_path = os.environ['LLAMA_SERVER_BIN_PATH']
  640. server_args = [
  641. '--host', context.server_fqdn,
  642. '--port', context.server_port,
  643. '--model', context.model_file
  644. ]
  645. if context.server_continuous_batching:
  646. server_args.append('--cont-batching')
  647. if context.server_embeddings:
  648. server_args.append('--embedding')
  649. if context.server_metrics:
  650. server_args.append('--metrics')
  651. if context.model_alias is not None:
  652. server_args.extend(['--alias', context.model_alias])
  653. if context.n_ctx is not None:
  654. server_args.extend(['--ctx-size', context.n_ctx])
  655. if context.n_slots is not None:
  656. server_args.extend(['--parallel', context.n_slots])
  657. if context.n_server_predict is not None:
  658. server_args.extend(['--n-predict', context.n_server_predict])
  659. if context.server_api_key is not None:
  660. server_args.extend(['--api-key', context.server_api_key])
  661. if context.debug:
  662. server_args.append('--verbose')
  663. print(f"starting server with: {context.server_path}", *server_args)
  664. context.server_process = subprocess.Popen(
  665. [str(arg) for arg in [context.server_path, *server_args]],
  666. close_fds=True)
  667. print(f"server pid={context.server_process.pid}")