a bgg$@sddlmZddlmZmZmZmZddlmZddl m Z ddl m Z ddl mZmZddlmZGdd d eeZGd d d eZGd d d e ZdS))Enum)AnyIteratorListOptional)CallbackManagerForLLMRun)LLM)GenerationChunk) BaseModel ConfigDict)enforce_stop_tokensc@seZdZdZdZdZdS)Devicez,The device to use for inference, cuda or cpucudacpuN)__name__ __module__ __qualname____doc__rrrrt/var/www/html/cobodadashboardai.evdpl.com/venv/lib/python3.9/site-packages/langchain_community/llms/titan_takeoff.pyr sr c@sfeZdZUdZeddZeed<ej Z eed<dZ eed<dZ e eed <d Zeed <d Zeed <dS) ReaderConfigzAConfiguration for the reader to be deployed in Titan Takeoff API.r)Zprotected_namespacesZ model_namedeviceprimaryconsumer_groupNtensor_parallelimax_seq_lengthmax_batch_size)rrrrr Z model_configstr__annotations__r rrrrrintrrrrrrrs    rcseZdZUdZdZeed<dZeed<dZ eed<dZ e ed <d Z e ed <ddddgfeeee eed fd d ZeedddZdeeeeeee edddZdeeeeeee eedddZZS) TitanTakeoffaTitan Takeoff API LLMs. Titan Takeoff is a wrapper to interface with Takeoff Inference API for generative text to text language models. You can use this wrapper to send requests to a generative language model and to deploy readers with Takeoff. Examples: This is an example how to deploy a generative language model and send requests. .. code-block:: python # Import the TitanTakeoff class from community package import time from langchain_community.llms import TitanTakeoff # Specify the embedding reader you'd like to deploy reader_1 = { "model_name": "TheBloke/Llama-2-7b-Chat-AWQ", "device": "cuda", "tensor_parallel": 1, "consumer_group": "llama" } # For every reader you pass into models arg Takeoff will spin # up a reader according to the specs you provide. If you don't # specify the arg no models are spun up and it assumes you have # already done this separately. llm = TitanTakeoff(models=[reader_1]) # Wait for the reader to be deployed, time needed depends on the # model size and your internet speed time.sleep(60) # Returns the query, ie a List[float], sent to `llama` consumer group # where we just spun up the Llama 7B model print(embed.invoke( "Where can I see football?", consumer_group="llama" )) # You can also send generation parameters to the model, any of the # following can be passed in as kwargs: # https://docs.titanml.co/docs/next/apis/Takeoff%20inference_REST_API/generate#request # for instance: print(embed.invoke( "Where can I see football?", consumer_group="llama", max_new_tokens=100 )) zhttp://localhostbase_urli porti mgmt_portF streamingNclient)r"r#r$r%modelscsptj||||dzddlm}Wnty>tdYn0||j|j|jd|_|D]}|j |qZdS)aInitialize the Titan Takeoff language wrapper. Args: base_url (str, optional): The base URL where the Takeoff Inference Server is listening. Defaults to `http://localhost`. port (int, optional): What port is Takeoff Inference API listening on. Defaults to 3000. mgmt_port (int, optional): What port is Takeoff Management API listening on. Defaults to 3001. streaming (bool, optional): Whether you want to by default use the generate_stream endpoint over generate to stream responses. Defaults to False. In reality, this is not significantly different as the streamed response is buffered and returned similar to the non-streamed response, but the run manager is applied per token generated. models (List[ReaderConfig], optional): Any readers you'd like to spin up on. Defaults to []. Raises: ImportError: If you haven't installed takeoff-client, you will get an ImportError. To remedy run `pip install 'takeoff-client==0.4.0'` )r"r#r$r%r) TakeoffClientzjtakeoff-client is required for TitanTakeoff. Please install it with `pip install 'takeoff-client>=0.4.0'`.)r#r$N) super__init__Ztakeoff_clientr( ImportErrorr"r#r$r&Z create_reader)selfr"r#r$r%r'r(model __class__rrr*os   zTitanTakeoff.__init__)returncCsdS)zReturn type of llm.Z titan_takeoffr)r,rrr _llm_typeszTitanTakeoff._llm_type)promptstop run_managerkwargsr0c Ks`|jr.d}|j|||dD]}||j7}q|S|jj|fi|}|d}|dur\t||}|S)aCall out to Titan Takeoff (Pro) generate endpoint. Args: prompt: The prompt to pass into the model. stop: Optional list of stop words to use when generating. run_manager: Optional callback manager to use when streaming. Returns: The string generated by the model. Example: .. code-block:: python model = TitanTakeoff() prompt = "What is the capital of the United Kingdom?" # Use of model(prompt), ie `__call__` was deprecated in LangChain 0.1.7, # use model.invoke(prompt) instead. response = model.invoke(prompt) )r2r3r4textN)r%_streamr7r&generater ) r,r2r3r4r5Z text_outputchunkresponser7rrr_calls   zTitanTakeoff._callc ks|jj|fi|}d}|D]|}||j7}d|vr|dr@d}t|dddkrn|dd\}} |d}|rt|d} d}|r|j| j d| Vq|rt| ddd} |r|j| j d| Vd S) aCall out to Titan Takeoff (Pro) stream endpoint. Args: prompt: The prompt to pass into the model. stop: Optional list of stop words to use when generating. run_manager: Optional callback manager to use when streaming. Yields: A dictionary like object containing a string token. Example: .. code-block:: python model = TitanTakeoff() prompt = "What is the capital of the United Kingdom?" response = model.stream(prompt) # OR model = TitanTakeoff(streaming=True) response = model.invoke(prompt) r6zdata: )r7)tokenzN) r&Zgenerate_streamdata startswithlensplitrstripr Zon_llm_new_tokenr7replace) r,r2r3r4r5r;bufferr7content_r:rrrr8s*     zTitanTakeoff._stream)NN)NN)rrrrr"rrr#r r$r%boolr&rrrr*propertyr1rrr<rr r8 __classcell__rrr.rr!-sL 2     .  1 r!N)enumrtypingrrrrZlangchain_core.callbacksrZ#langchain_core.language_models.llmsrZlangchain_core.outputsr Zpydanticr r Zlangchain_community.llms.utilsr rr rr!rrrrs