Spaces:

iblfe
/

test

Runtime error

App Files Files Community

test / gradio_utils /grclient.py

iblfe

Upload folder using huggingface_hub

b585c7f verified 5 months ago

raw history blame contribute delete

No virus

41.4 kB

	from __future__ import annotations

	import difflib
	import traceback
	import concurrent.futures
	import os
	import concurrent.futures
	import time
	import urllib.parse
	import uuid
	from concurrent.futures import Future
	from datetime import timedelta
	from enum import Enum
	from pathlib import Path
	from typing import Callable, Generator, Any, Union, List
	import ast
	from packaging import version

	os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1"

	from huggingface_hub import SpaceStage
	from huggingface_hub.utils import (
	build_hf_headers,
	)

	from gradio_client import utils

	from importlib.metadata import distribution, PackageNotFoundError

	try:
	assert distribution('gradio_client') is not None
	have_gradio_client = True
	is_gradio_client_version7 = distribution('gradio_client').version.startswith('0.7.')
	except (PackageNotFoundError, AssertionError):
	have_gradio_client = False
	is_gradio_client_version7 = False


	from gradio_client.client import Job, DEFAULT_TEMP_DIR, Endpoint
	from gradio_client import Client


	def check_job(job, timeout=0.0, raise_exception=True, verbose=False):
	if timeout == 0:
	e = job.future._exception
	else:
	try:
	e = job.future.exception(timeout=timeout)
	except concurrent.futures.TimeoutError:
	# not enough time to determine
	if verbose:
	print("not enough time to determine job status: %s" % timeout)
	e = None
	if e:
	# raise before complain about empty response if some error hit
	if raise_exception:
	raise RuntimeError(e)
	else:
	return e


	# Local copy of minimal version from h2oGPT server
	class LangChainAction(Enum):
	"""LangChain action"""

	QUERY = "Query"
	SUMMARIZE_MAP = "Summarize"
	EXTRACT = "Extract"


	pre_prompt_query0 = "Pay attention and remember the information below, which will help to answer the question or imperative after the context ends."
	prompt_query0 = "According to only the information in the document sources provided within the context above: "

	pre_prompt_summary0 = """"""
	prompt_summary0 = "Using only the information in the document sources above, write a condensed and concise summary of key results (preferably as bullet points)."

	pre_prompt_extraction0 = """In order to extract information, pay attention to the following text."""
	prompt_extraction0 = "Using only the information in the document sources above, extract "


	hyde_llm_prompt0 = "Answer this question with vibrant details in order for some NLP embedding model to use that answer as better query than original question: "


	class GradioClient(Client):
	"""
	Parent class of gradio client
	To handle automatically refreshing client if detect gradio server changed
	"""

	def __init__(
	self,
	src: str,
	hf_token: str \| None = None,
	max_workers: int = 40,
	serialize: bool = None,
	output_dir: str \| Path \| None = DEFAULT_TEMP_DIR,
	verbose: bool = False,
	auth: tuple[str, str] \| None = None,
	h2ogpt_key: str = None,
	persist: bool = False,
	check_hash: bool = True,
	check_model_name: bool = False,
	):
	"""
	Parameters:
	src: Either the name of the Hugging Face Space to load, (e.g. "abidlabs/whisper-large-v2") or the full URL (including "http" or "https") of the hosted Gradio app to load (e.g. "http://mydomain.com/app" or "https://bec81a83-5b5c-471e.gradio.live/").
	hf_token: The Hugging Face token to use to access private Spaces. Automatically fetched if you are logged in via the Hugging Face Hub CLI. Obtain from: https://huggingface.co/settings/token
	max_workers: The maximum number of thread workers that can be used to make requests to the remote Gradio app simultaneously.
	serialize: Whether the client should serialize the inputs and deserialize the outputs of the remote API. If set to False, the client will pass the inputs and outputs as-is, without serializing/deserializing them. E.g. you if you set this to False, you'd submit an image in base64 format instead of a filepath, and you'd get back an image in base64 format from the remote API instead of a filepath.
	output_dir: The directory to save files that are downloaded from the remote API. If None, reads from the GRADIO_TEMP_DIR environment variable. Defaults to a temporary directory on your machine.
	verbose: Whether the client should print statements to the console.

	h2ogpt_key: h2oGPT key to gain access to the server
	persist: whether to persist the state, so repeated calls are aware of the prior user session
	This allows the scratch MyData to be reused, etc.
	This also maintains the chat_conversation history
	check_hash: whether to check git hash for consistency between server and client to ensure API always up to date
	check_model_name: whether to check the model name here (adds delays), or just let server fail (fater)
	"""
	if serialize is None:
	# else converts inputs arbitrarily and outputs mutate
	# False keeps as-is and is normal for h2oGPT
	serialize = False
	self.args = tuple([src])
	self.kwargs = dict(
	hf_token=hf_token,
	max_workers=max_workers,
	serialize=serialize,
	output_dir=output_dir,
	verbose=verbose,
	h2ogpt_key=h2ogpt_key,
	persist=persist,
	check_hash=check_hash,
	check_model_name=check_model_name,
	)
	if is_gradio_client_version7:
	self.kwargs.update(dict(auth=auth))

	self.verbose = verbose
	self.hf_token = hf_token
	self.serialize = serialize
	self.space_id = None
	self.cookies: dict[str, str] = {}
	if is_gradio_client_version7:
	self.output_dir = (
	str(output_dir) if isinstance(output_dir, Path) else output_dir
	)
	else:
	self.output_dir = output_dir
	self.max_workers = max_workers
	self.src = src
	self.auth = auth
	self.config = None
	self.h2ogpt_key = h2ogpt_key
	self.persist = persist
	self.check_hash = check_hash
	self.check_model_name = check_model_name

	self.chat_conversation = [] # internal for persist=True
	self.server_hash = None # internal

	def __repr__(self):
	if self.config:
	return self.view_api(print_info=False, return_format="str")
	return "Not setup for %s" % self.src

	def __str__(self):
	if self.config:
	return self.view_api(print_info=False, return_format="str")
	return "Not setup for %s" % self.src

	def setup(self):
	src = self.src

	self.headers = build_hf_headers(
	token=self.hf_token,
	library_name="gradio_client",
	library_version=utils.__version__,
	)
	# self.headers.pop('authorization', None) # else get illegal Bearer for old servers
	if src.startswith("http://") or src.startswith("https://"):
	_src = src if src.endswith("/") else src + "/"
	else:
	_src = self._space_name_to_src(src)
	if _src is None:
	raise ValueError(
	f"Could not find Space: {src}. If it is a private Space, please provide an hf_token."
	)
	self.space_id = src
	self.src = _src
	state = self._get_space_state()
	if state == SpaceStage.BUILDING:
	if self.verbose:
	print("Space is still building. Please wait...")
	while self._get_space_state() == SpaceStage.BUILDING:
	time.sleep(2) # so we don't get rate limited by the API
	pass
	if state in utils.INVALID_RUNTIME:
	raise ValueError(
	f"The current space is in the invalid state: {state}. "
	"Please contact the owner to fix this."
	)
	if self.verbose:
	print(f"Loaded as API: {self.src} ✔")

	if is_gradio_client_version7:
	if self.auth is not None:
	self._login(self.auth)

	self.api_url = urllib.parse.urljoin(self.src, utils.API_URL)
	if is_gradio_client_version7:
	self.sse_url = urllib.parse.urljoin(self.src, utils.SSE_URL)
	self.sse_data_url = urllib.parse.urljoin(self.src, utils.SSE_DATA_URL)
	self.ws_url = urllib.parse.urljoin(
	self.src.replace("http", "ws", 1), utils.WS_URL
	)
	self.upload_url = urllib.parse.urljoin(self.src, utils.UPLOAD_URL)
	self.reset_url = urllib.parse.urljoin(self.src, utils.RESET_URL)
	self.config = self._get_config()
	if is_gradio_client_version7:
	self.protocol: str = self.config.get("protocol", "ws")
	self.app_version = version.parse(self.config.get("version", "2.0"))
	self._info = self._get_api_info()
	self.session_hash = str(uuid.uuid4())

	if is_gradio_client_version7:
	from gradio_client.client import EndpointV3Compatibility
	endpoint_class = (
	Endpoint if self.protocol.startswith("sse") else EndpointV3Compatibility
	)
	else:
	endpoint_class = Endpoint

	if is_gradio_client_version7:
	self.endpoints = [
	endpoint_class(self, fn_index, dependency, self.protocol)
	for fn_index, dependency in enumerate(self.config["dependencies"])
	]
	else:
	self.endpoints = [
	endpoint_class(self, fn_index, dependency)
	for fn_index, dependency in enumerate(self.config["dependencies"])
	]

	# Create a pool of threads to handle the requests
	self.executor = concurrent.futures.ThreadPoolExecutor(
	max_workers=self.max_workers
	)

	# Disable telemetry by setting the env variable HF_HUB_DISABLE_TELEMETRY=1
	# threading.Thread(target=self._telemetry_thread).start()

	self.server_hash = self.get_server_hash()

	if is_gradio_client_version7:
	self.stream_open = False
	self.streaming_future: Future \| None = None
	from gradio_client.utils import Message
	self.pending_messages_per_event: dict[str, list[Message \| None]] = {}
	self.pending_event_ids: set[str] = set()

	return self

	def get_server_hash(self):
	if self.config is None:
	self.setup()
	"""
	Get server hash using super without any refresh action triggered
	Returns: git hash of gradio server
	"""
	if self.check_hash:
	return super().submit(api_name="/system_hash").result()
	else:
	return "GET_GITHASH"

	def refresh_client_if_should(self):
	if self.config is None:
	self.setup()
	# get current hash in order to update api_name -> fn_index map in case gradio server changed
	# FIXME: Could add cli api as hash
	server_hash = self.get_server_hash()
	if self.server_hash != server_hash:
	if self.verbose:
	print("server hash changed: %s %s" % (self.server_hash, server_hash), flush=True)
	if self.server_hash is not None and self.persist:
	if self.verbose:
	print("Failed to persist due to server hash change, only kept chat_conversation not user session hash", flush=True)
	# risky to persist if hash changed
	self.refresh_client()
	self.server_hash = server_hash

	def refresh_client(self):
	"""
	Ensure every client call is independent
	Also ensure map between api_name and fn_index is updated in case server changed (e.g. restarted with new code)
	Returns:
	"""
	if self.config is None:
	self.setup()

	kwargs = self.kwargs.copy()
	kwargs.pop('h2ogpt_key', None)
	kwargs.pop('persist', None)
	kwargs.pop('check_hash', None)
	kwargs.pop('check_model_name', None)
	ntrials = 3
	client = None
	for trial in range(0, ntrials + 1):
	try:
	client = Client(self.args, *kwargs)
	except ValueError as e:
	if trial >= ntrials:
	raise
	else:
	if self.verbose:
	print("Trying refresh %d/%d %s" % (trial, ntrials - 1, str(e)))
	trial += 1
	time.sleep(10)
	if client is None:
	raise RuntimeError("Failed to get new client")
	session_hash0 = self.session_hash if self.persist else None
	for k, v in client.__dict__.items():
	setattr(self, k, v)
	if session_hash0:
	# keep same system hash in case server API only changed and not restarted
	self.session_hash = session_hash0
	if self.verbose:
	print("Hit refresh_client(): %s %s" % (self.session_hash, session_hash0))
	# ensure server hash also updated
	self.server_hash = self.get_server_hash()

	def clone(self):
	if self.config is None:
	self.setup()
	client = GradioClient("")
	for k, v in self.__dict__.items():
	setattr(client, k, v)
	client.reset_session()
	client.executor = concurrent.futures.ThreadPoolExecutor(
	max_workers=self.max_workers
	)
	client.endpoints = [
	Endpoint(client, fn_index, dependency)
	for fn_index, dependency in enumerate(client.config["dependencies"])
	]
	# transfer internals in case used
	client.server_hash = self.server_hash
	client.chat_conversation = self.chat_conversation
	return client

	def submit(
	self,
	*args,
	api_name: str \| None = None,
	fn_index: int \| None = None,
	result_callbacks: Callable \| list[Callable] \| None = None,
	) -> Job:
	if self.config is None:
	self.setup()
	# Note predict calls submit
	try:
	self.refresh_client_if_should()
	job = super().submit(*args, api_name=api_name, fn_index=fn_index)
	except Exception as e:
	print("Hit e=%s\n\n%s" % (str(e), traceback.format_exc()), flush=True)
	# force reconfig in case only that
	self.refresh_client()
	job = super().submit(*args, api_name=api_name, fn_index=fn_index)

	# see if immediately failed
	e = check_job(job, timeout=0.01, raise_exception=False)
	if e is not None:
	print(
	"GR job failed: %s %s"
	% (str(e), "".join(traceback.format_tb(e.__traceback__))),
	flush=True,
	)
	# force reconfig in case only that
	self.refresh_client()
	job = super().submit(*args, api_name=api_name, fn_index=fn_index)
	e2 = check_job(job, timeout=0.1, raise_exception=False)
	if e2 is not None:
	print(
	"GR job failed again: %s\n%s"
	% (str(e2), "".join(traceback.format_tb(e2.__traceback__))),
	flush=True,
	)

	return job

	def question(self, instruction, args, *kwargs) -> str:
	"""
	Prompt LLM (direct to LLM with instruct prompting required for instruct models) and get response
	"""
	kwargs["instruction"] = kwargs.get("instruction", instruction)
	kwargs["langchain_action"] = LangChainAction.QUERY.value
	kwargs["langchain_mode"] = 'LLM'
	ret = ''
	for response, texts_out in self.query_or_summarize_or_extract(args, *kwargs):
	ret = response
	return ret

	def question_stream(self, instruction, args, *kwargs) -> str:
	"""
	Prompt LLM (direct to LLM with instruct prompting required for instruct models) and get response
	"""
	kwargs["instruction"] = kwargs.get("instruction", instruction)
	kwargs["langchain_action"] = LangChainAction.QUERY.value
	kwargs["langchain_mode"] = 'LLM'
	ret = yield from self.query_or_summarize_or_extract(args, *kwargs)
	return ret

	def query(self, query, args, *kwargs) -> str:
	"""
	Search for documents matching a query, then ask that query to LLM with those documents
	"""
	kwargs["instruction"] = kwargs.get("instruction", query)
	kwargs["langchain_action"] = LangChainAction.QUERY.value
	ret = ''
	for response, texts_out in self.query_or_summarize_or_extract(args, *kwargs):
	ret = response
	return ret

	def query_stream(self, query, args, *kwargs) -> Generator[tuple[str \| list[str], list[str]], None, None]:
	"""
	Search for documents matching a query, then ask that query to LLM with those documents
	"""
	kwargs["instruction"] = kwargs.get("instruction", query)
	kwargs["langchain_action"] = LangChainAction.QUERY.value
	ret = yield from self.query_or_summarize_or_extract(args, *kwargs)
	return ret

	def summarize(self, args, query=None, focus=None, *kwargs) -> str:
	"""
	Search for documents matching a focus, then ask a query to LLM with those documents
	If focus "" or None, no similarity search is done and all documents (up to top_k_docs) are used
	"""
	kwargs["prompt_summary"] = kwargs.get("prompt_summary", query or prompt_summary0)
	kwargs["instruction"] = kwargs.get('instruction', focus)
	kwargs["langchain_action"] = LangChainAction.SUMMARIZE_MAP.value
	ret = ''
	for response, texts_out in self.query_or_summarize_or_extract(args, *kwargs):
	ret = response
	return ret

	def summarize_stream(self, args, query=None, focus=None, *kwargs) -> str:
	"""
	Search for documents matching a focus, then ask a query to LLM with those documents
	If focus "" or None, no similarity search is done and all documents (up to top_k_docs) are used
	"""
	kwargs["prompt_summary"] = kwargs.get("prompt_summary", query or prompt_summary0)
	kwargs["instruction"] = kwargs.get('instruction', focus)
	kwargs["langchain_action"] = LangChainAction.SUMMARIZE_MAP.value
	ret = yield from self.query_or_summarize_or_extract(args, *kwargs)
	return ret

	def extract(self, args, query=None, focus=None, *kwargs) -> list[str]:
	"""
	Search for documents matching a focus, then ask a query to LLM with those documents
	If focus "" or None, no similarity search is done and all documents (up to top_k_docs) are used
	"""
	kwargs["prompt_extraction"] = kwargs.get("prompt_extraction", query or prompt_extraction0)
	kwargs["instruction"] = kwargs.get('instruction', focus)
	kwargs["langchain_action"] = LangChainAction.EXTRACT.value
	ret = ''
	for response, texts_out in self.query_or_summarize_or_extract(args, *kwargs):
	ret = response
	return ret

	def extract_stream(self, args, query=None, focus=None, *kwargs) -> list[str]:
	"""
	Search for documents matching a focus, then ask a query to LLM with those documents
	If focus "" or None, no similarity search is done and all documents (up to top_k_docs) are used
	"""
	kwargs["prompt_extraction"] = kwargs.get("prompt_extraction", query or prompt_extraction0)
	kwargs["instruction"] = kwargs.get('instruction', focus)
	kwargs["langchain_action"] = LangChainAction.EXTRACT.value
	ret = yield from self.query_or_summarize_or_extract(args, *kwargs)
	return ret

	def query_or_summarize_or_extract(self,
	h2ogpt_key: str = None,

	instruction: str = "",

	text: list[str] \| str \| None = None,
	file: list[str] \| str \| None = None,
	url: list[str] \| str \| None = None,
	embed: bool = True,
	chunk: bool = True,
	chunk_size: int = 512,

	langchain_mode: str = None,
	langchain_action: str \| None = None,
	langchain_agents: List[str] = [],
	top_k_docs: int = 10,
	document_choice: Union[str, List[str]] = "All",
	document_subset: str = "Relevant",
	document_source_substrings: Union[str, List[str]] = [],
	document_source_substrings_op: str = 'and',
	document_content_substrings: Union[str, List[str]] = [],
	document_content_substrings_op: str = 'and',

	system_prompt: str \| None = '',
	pre_prompt_query: str \| None = pre_prompt_query0,
	prompt_query: str \| None = prompt_query0,
	pre_prompt_summary: str \| None = pre_prompt_summary0,
	prompt_summary: str \| None = prompt_summary0,
	pre_prompt_extraction: str \| None = pre_prompt_extraction0,
	prompt_extraction: str \| None = prompt_extraction0,
	hyde_llm_prompt: str \| None = hyde_llm_prompt0,

	model: str \| int \| None = None,
	stream_output: bool = False,
	do_sample: bool = False,
	temperature: float = 0.0,
	top_p: float = 0.75,
	top_k: int = 40,
	repetition_penalty: float = 1.07,
	penalty_alpha: float = 0.0,
	max_time: int = 360,
	max_new_tokens: int = 1024,

	add_search_to_context: bool = False,
	chat_conversation: list[tuple[str, str]] \| None = None,
	text_context_list: list[str] \| None = None,
	docs_ordering_type: str \| None = None,
	min_max_new_tokens: int = 512,
	max_input_tokens: int = -1,
	max_total_input_tokens: int = -1,
	docs_token_handling: str = "split_or_merge",
	docs_joiner: str = "\n\n",
	hyde_level: int = 0,
	hyde_template: str = None,
	hyde_show_only_final: bool = True,
	doc_json_mode: bool = False,

	asserts: bool = False,
	) -> Generator[tuple[str \| list[str], list[str]], None, None]:
	"""
	Query or Summarize or Extract using h2oGPT
	Args:
	instruction: Query for LLM chat. Used for similarity search

	For query, prompt template is:
	"{pre_prompt_query}
	\"\"\"
	{content}
	\"\"\"
	{prompt_query}{instruction}"
	If added to summarization, prompt template is
	"{pre_prompt_summary}
	\"\"\"
	{content}
	\"\"\"
	Focusing on {instruction}, {prompt_summary}"
	text: textual content or list of such contents
	file: a local file to upload or files to upload
	url: a url to give or urls to use
	embed: whether to embed content uploaded

	langchain_mode: "LLM" to talk to LLM with no docs, "MyData" for personal docs, "UserData" for shared docs, etc.
	langchain_action: Action to take, "Query" or "Summarize" or "Extract"
	langchain_agents: Which agents to use, if any
	top_k_docs: number of document parts.
	When doing query, number of chunks
	When doing summarization, not related to vectorDB chunks that are not used
	E.g. if PDF, then number of pages
	chunk: whether to chunk sources for document Q/A
	chunk_size: Size in characters of chunks
	document_choice: Which documents ("All" means all) -- need to use upload_api API call to get server's name if want to select
	document_subset: Type of query, see src/gen.py
	document_source_substrings: See gen.py
	document_source_substrings_op: See gen.py
	document_content_substrings: See gen.py
	document_content_substrings_op: See gen.py

	system_prompt: pass system prompt to models that support it.
	If 'auto' or None, then use automatic version
	If '', then use no system prompt (default)
	pre_prompt_query: Prompt that comes before document part
	prompt_query: Prompt that comes after document part
	pre_prompt_summary: Prompt that comes before document part
	None makes h2oGPT internally use its defaults
	E.g. "In order to write a concise single-paragraph or bulleted list summary, pay attention to the following text"
	prompt_summary: Prompt that comes after document part
	None makes h2oGPT internally use its defaults
	E.g. "Using only the text above, write a condensed and concise summary of key results (preferably as bullet points):\n"
	i.e. for some internal document part fstring, the template looks like:
	template = "%s
	\"\"\"
	%s
	\"\"\"
	%s" % (pre_prompt_summary, fstring, prompt_summary)
	hyde_llm_prompt: hyde prompt for first step when using LLM
	h2ogpt_key: Access Key to h2oGPT server (if not already set in client at init time)
	model: base_model name or integer index of model_lock on h2oGPT server
	None results in use of first (0th index) model in server
	to get list of models do client.list_models()
	pre_prompt_extraction: Same as pre_prompt_summary but for when doing extraction
	prompt_extraction: Same as prompt_summary but for when doing extraction
	do_sample: see src/gen.py
	temperature: see src/gen.py
	top_p: see src/gen.py
	top_k: see src/gen.py
	repetition_penalty: see src/gen.py
	penalty_alpha: see src/gen.py
	max_new_tokens: see src/gen.py
	min_max_new_tokens: see src/gen.py
	max_input_tokens: see src/gen.py
	max_total_input_tokens: see src/gen.py

	stream_output: Whether to stream output
	do_sample: whether to sample
	max_time: how long to take

	add_search_to_context: Whether to do web search and add results to context
	chat_conversation: List of tuples for (human, bot) conversation that will be pre-appended to an (instruction, None) case for a query
	text_context_list: List of strings to add to context for non-database version of document Q/A for faster handling via API etc.
	Forces LangChain code path and uses as many entries in list as possible given max_seq_len, with first assumed to be most relevant and to go near prompt.
	docs_ordering_type: By default uses 'reverse_ucurve_sort' for optimal retrieval
	max_input_tokens: Max input tokens to place into model context for each LLM call
	-1 means auto, fully fill context for query, and fill by original document chunk for summarization
	>=0 means use that to limit context filling to that many tokens
	max_total_input_tokens: like max_input_tokens but instead of per LLM call, applies across all LLM calls for single summarization/extraction action
	max_new_tokens: Maximum new tokens
	min_max_new_tokens: minimum value for max_new_tokens when auto-adjusting for content of prompt, docs, etc.

	docs_token_handling: 'chunk' means fill context with top_k_docs (limited by max_input_tokens or model_max_len) chunks for query
	or top_k_docs original document chunks summarization
	None or 'split_or_merge' means same as 'chunk' for query, while for summarization merges documents to fill up to max_input_tokens or model_max_len tokens
	docs_joiner: string to join lists of text when doing split_or_merge. None means '\n\n'
	hyde_level: 0-3 for HYDE.
	0 uses just query to find similarity with docs
	1 uses query + pure LLM response to find similarity with docs
	2: uses query + LLM response using docs to find similarity with docs
	3+: etc.
	hyde_template: see src/gen.py
	hyde_show_only_final: see src/gen.py
	doc_json_mode: see src/gen.py

	asserts: whether to do asserts to ensure handling is correct

	Returns: summary/answer: str or extraction List[str]

	"""
	if self.config is None:
	self.setup()
	if self.persist:
	client = self
	else:
	client = self.clone()
	h2ogpt_key = h2ogpt_key or self.h2ogpt_key
	client.h2ogpt_key = h2ogpt_key

	self.check_model(model)

	# chunking not used here
	# MyData specifies scratch space, only persisted for this individual client call
	langchain_mode = langchain_mode or "MyData"
	loaders = tuple([None, None, None, None, None, None])
	doc_options = tuple([langchain_mode, chunk, chunk_size, embed])
	asserts \|= bool(os.getenv("HARD_ASSERTS", False))
	if (
	text
	and isinstance(text, list)
	and not file
	and not url
	and not text_context_list
	):
	# then can do optimized text-only path
	text_context_list = text
	text = None

	res = []
	if text:
	t0 = time.time()
	res = client.predict(
	text, doc_options, loaders, h2ogpt_key, api_name="/add_text"
	)
	t1 = time.time()
	print("upload text: %s" % str(timedelta(seconds=t1 - t0)), flush=True)
	if asserts:
	assert res[0] is None
	assert res[1] == langchain_mode
	assert "user_paste" in res[2]
	assert res[3] == ""
	if file:
	# upload file(s). Can be list or single file
	# after below call, "file" replaced with remote location of file
	_, file = client.predict(file, api_name="/upload_api")

	res = client.predict(
	file, doc_options, loaders, h2ogpt_key, api_name="/add_file_api"
	)
	if asserts:
	assert res[0] is None
	assert res[1] == langchain_mode
	assert os.path.basename(file) in res[2]
	assert res[3] == ""
	if url:
	res = client.predict(
	url, doc_options, loaders, h2ogpt_key, api_name="/add_url"
	)
	if asserts:
	assert res[0] is None
	assert res[1] == langchain_mode
	assert url in res[2]
	assert res[3] == ""
	assert res[4] # should have file name or something similar
	if res and not res[4] and "Exception" in res[2]:
	print("Exception: %s" % res[2], flush=True)

	# ask for summary, need to use same client if using MyData
	api_name = "/submit_nochat_api" # NOTE: like submit_nochat but stable API for string dict passing

	pre_prompt_summary = pre_prompt_summary \
	if langchain_action == LangChainAction.SUMMARIZE_MAP.value \
	else pre_prompt_extraction
	prompt_summary = prompt_summary \
	if langchain_action == LangChainAction.SUMMARIZE_MAP.value \
	else prompt_extraction

	kwargs = dict(
	h2ogpt_key=h2ogpt_key,

	instruction=instruction,

	langchain_mode=langchain_mode,
	langchain_action=langchain_action, # uses full document, not vectorDB chunks
	langchain_agents=langchain_agents,
	top_k_docs=top_k_docs,
	document_choice=document_choice,
	document_subset=document_subset,
	document_source_substrings=document_source_substrings,
	document_source_substrings_op=document_source_substrings_op,
	document_content_substrings=document_content_substrings,
	document_content_substrings_op=document_content_substrings_op,

	system_prompt=system_prompt,
	pre_prompt_query=pre_prompt_query,
	prompt_query=prompt_query,
	pre_prompt_summary=pre_prompt_summary,
	prompt_summary=prompt_summary,
	hyde_llm_prompt=hyde_llm_prompt,

	visible_models=model,
	stream_output=stream_output,
	do_sample=do_sample,
	temperature=temperature,
	top_p=top_p,
	top_k=top_k,
	repetition_penalty=repetition_penalty,
	penalty_alpha=penalty_alpha,
	max_time=max_time,
	max_new_tokens=max_new_tokens,

	add_search_to_context=add_search_to_context,
	chat_conversation=chat_conversation if chat_conversation else self.chat_conversation,
	text_context_list=text_context_list,
	docs_ordering_type=docs_ordering_type,
	min_max_new_tokens=min_max_new_tokens,
	max_input_tokens=max_input_tokens,
	max_total_input_tokens=max_total_input_tokens,
	docs_token_handling=docs_token_handling,
	docs_joiner=docs_joiner,
	hyde_level=hyde_level,
	hyde_template=hyde_template,
	hyde_show_only_final=hyde_show_only_final,
	doc_json_mode=doc_json_mode,
	)

	# in case server changed, update in case clone()
	self.server_hash = client.server_hash

	# ensure can fill conversation
	self.chat_conversation.append((instruction, None))

	# get result
	trials = 3
	for trial in range(trials):
	try:
	if not stream_output:
	res = client.predict(
	str(dict(kwargs)),
	api_name=api_name,
	)
	# in case server changed, update in case clone()
	self.server_hash = client.server_hash
	res = ast.literal_eval(res)
	response = res["response"]
	if langchain_action != LangChainAction.EXTRACT.value:
	response = response.strip()
	else:
	response = [r.strip() for r in ast.literal_eval(response)]
	sources = res["sources"]
	scores_out = [x["score"] for x in sources]
	texts_out = [x["content"] for x in sources]
	if asserts:
	if text and not file and not url:
	assert any(
	text[:cutoff] == texts_out for cutoff in range(len(text))
	)
	assert len(texts_out) == len(scores_out)

	yield response, texts_out
	self.chat_conversation[-1] = (instruction, response)
	else:
	job = client.submit(str(dict(kwargs)), api_name=api_name)
	text0 = ""
	response = ""
	texts_out = []
	while not job.done():
	if job.communicator.job.latest_status.code.name == "FINISHED":
	break
	e = check_job(job, timeout=0, raise_exception=False)
	if e is not None:
	break
	outputs_list = job.communicator.job.outputs
	if outputs_list:
	res = job.communicator.job.outputs[-1]
	res_dict = ast.literal_eval(res)
	response = res_dict["response"] # keeps growing
	text_chunk = response[len(text0):] # only keep new stuff
	if not text_chunk:
	time.sleep(0.001)
	continue
	text0 = response
	assert text_chunk, "must yield non-empty string"
	yield text_chunk, texts_out
	time.sleep(
	0.1
	) # let LLM deliver larger chunks, don't need to get every token output immediately

	# Get final response (if anything left), but also get the actual references (texts_out), above is empty.
	res_all = job.outputs()
	if len(res_all) > 0:
	# 0.1 slightly longer than 0.02 in open source
	check_job(job, timeout=0.1, raise_exception=True)

	res = res_all[-1]
	res_dict = ast.literal_eval(res)
	response = res_dict["response"]
	sources = res_dict["sources"]
	texts_out = [x["content"] for x in sources]
	yield response[len(text0):], texts_out
	self.chat_conversation[-1] = (instruction, response[len(text0):])
	else:
	# 1.0 slightly longer than 0.3 in open source
	check_job(job, timeout=1.0, raise_exception=True)
	yield response[len(text0):], texts_out
	self.chat_conversation[-1] = (instruction, response[len(text0):])
	break
	except Exception as e:
	print(
	"h2oGPT predict failed: %s %s"
	% (str(e), "".join(traceback.format_tb(e.__traceback__))),
	flush=True,
	)
	if trial == trials - 1:
	raise
	else:
	print("trying again: %s" % trial, flush=True)
	time.sleep(1 * trial)
	finally:
	# in case server changed, update in case clone()
	self.server_hash = client.server_hash

	def check_model(self, model):
	if model != 0 and self.check_model_name:
	valid_llms = self.list_models()
	if (
	isinstance(model, int)
	and model >= len(valid_llms)
	or isinstance(model, str)
	and model not in valid_llms
	):
	did_you_mean = ""
	if isinstance(model, str):
	alt = difflib.get_close_matches(model, valid_llms, 1)
	if alt:
	did_you_mean = f"\nDid you mean {repr(alt[0])}?"
	raise RuntimeError(
	f"Invalid llm: {repr(model)}, must be either an integer between "
	f"0 and {len(valid_llms) - 1} or one of the following values: {valid_llms}.{did_you_mean}"
	)

	def get_models_full(self) -> list[dict[str, Any]]:
	"""
	Full model info in list if dict
	"""
	if self.config is None:
	self.setup()
	return ast.literal_eval(self.predict(api_name="/model_names"))

	def list_models(self) -> list[str]:
	"""
	Model names available from endpoint
	"""
	if self.config is None:
	self.setup()
	return [x['base_model'] for x in ast.literal_eval(self.predict(api_name="/model_names"))]