dh! SSKrSSKrSSKrSSKJrJrJrJrJ r SSK J r SSK J r SSKJr SSKJr \R$"\5rSS.S\S \S \S \ \\S \S \4 SjjrS\S\S \4Sjr"SS\ 5rg)N)AnyCallableListMappingOptional)CallbackManagerForLLMRun)LLM) ConfigDictenforce_stop_tokens)stoppipelinepromptargsr kwargsreturnc:U"U/UQ70UD6nUb [XR5nU$)zInference function to send to the remote hardware. Accepts a pipeline callable (or, more likely, a key pointing to the model on the cluster's object store) and returns text predictions for each document in the batch. r )rrr rrtexts \/var/www/html/shao/venv/lib/python3.13/site-packages/langchain_community/llms/self_hosted.py_generate_textrs- F ,T ,V ,D "4. Kdevicec"[U[5(a+[US5n[R"U5nSSS5 [ R RS5bSSKnURR5nUS:dX:a[SUSUS35eUS:aUS:a[RS U5 URU5UlURR!UR5UlU$!,(df  N=f) z+Send a pipeline to a device on the cluster.rbNtorchrz Got device==z', device is required to be within [-1, )zDevice has %d GPUs available. Provide device={deviceId} to `from_model_id` to use availableGPUs for execution. deviceId is -1 for CPU and can be a positive integer associated with CUDA device id.) isinstancestropenpickleload importlibutil find_specrcuda device_count ValueErrorloggerwarningrmodelto)rrfrcuda_device_counts r_send_pipeline_to_devicer/#s(C  (D !Q{{1~H" ~~(4!JJ335 B;66vh'88I7J!M  A:+a/ NNL"   ,,v.!**8??; O3" !s D Dc|^\rSrSr%SrSr\\S'Sr\\S'\ r \ \S'Sr \\S'\ \S'Sr \\\S 'S S /r\\\S 'S r\\S'\"SS9rS\4U4Sjjr\SS\S\S \\\S\S\S\4 Sjj5r\S\\\44Sj5r\S\4Sj5rSS\S\\\S\\ S\S\4 Sjjr!Sr"U=r#$) SelfHostedPipelineBaModel inference on self-hosted remote hardware. Supported hardware includes auto-launched instances on AWS, GCP, Azure, and Lambda, as well as servers specified by IP address and SSH credentials (such as on-prem, or another cloud like Paperspace, Coreweave, etc.). To use, you should have the ``runhouse`` python package installed. Example for custom pipeline and inference functions: .. code-block:: python from langchain_community.llms import SelfHostedPipeline from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline import runhouse as rh def load_pipeline(): tokenizer = AutoTokenizer.from_pretrained("gpt2") model = AutoModelForCausalLM.from_pretrained("gpt2") return pipeline( "text-generation", model=model, tokenizer=tokenizer, max_new_tokens=10 ) def inference_fn(pipeline, prompt, stop = None): return pipeline(prompt)[0]["generated_text"] gpu = rh.cluster(name="rh-a10x", instance_type="A100:1") llm = SelfHostedPipeline( model_load_fn=load_pipeline, hardware=gpu, model_reqs=model_reqs, inference_fn=inference_fn ) Example for <2GB model (can be serialized and sent directly to the server): .. code-block:: python from langchain_community.llms import SelfHostedPipeline import runhouse as rh gpu = rh.cluster(name="rh-a10x", instance_type="A100:1") my_model = ... llm = SelfHostedPipeline.from_pipeline( pipeline=my_model, hardware=gpu, model_reqs=["./", "torch", "transformers"], ) Example passing model path for larger models: .. code-block:: python from langchain_community.llms import SelfHostedPipeline import runhouse as rh import pickle from transformers import pipeline generator = pipeline(model="gpt2") rh.blob(pickle.dumps(generator), path="models/pipeline.pkl" ).save().to(gpu, path="models") llm = SelfHostedPipeline.from_pipeline( pipeline="models/pipeline.pkl", hardware=gpu, model_reqs=["./", "torch", "transformers"], ) N pipeline_refclient inference_fnhardware model_load_fnload_fn_kwargsz./r model_reqsFallow_dangerous_deserializationforbid)extrarc >URS5(d [S5e[TU] "S0UD6 SSKnUR URS9RURURS9nUR=(d 0nUR"S0UD6Ul UR URS9RURURS9Ulg![ a [ S5ef=f) zInit the pipeline with an auxiliary function. The load function must be in global scope to be imported and run on the server, i.e. in a module and not a REPL or closure. Then, initialize the remote inference function. r:aQSelfHostedPipeline relies on the pickle module. You will need to set allow_dangerous_deserialization=True if you want to opt-in to allow deserialization of data using pickle.Data can be compromised by a malicious actor if not handled properly to include a malicious payload that when deserialized with pickle can execute arbitrary code. rNzXCould not import runhouse python package. Please install it with `pip install runhouse`.)fn)reqs)getr(super__init__runhouse ImportErrorfunctionr7r,r6r9r8remoter3r5r4)selfrrhremote_load_fn_load_fn_kwargs __class__s rrCSelfHostedPipeline.__init__szz;<<6  "6"  !(:(:;>> MM? --3*11DODkkT%6%6k7:: MM;   A  s C C6rrrc [U[5(d[RS5 XS.nU"SU[USS/U=(d /-S.UD6$)z=Init the SelfHostedPipeline from a pipeline object or string.zSerializing pipeline to send to remote hardware. Note, it can be quite slowto serialize and send large models with each execution. Consider sending the pipelineto the cluster and passing the path to the pipeline instead.)rr transformersr)r8r7r6r9r@)rrr)r*r/)clsrr6r9rrr8s r from_pipeline SelfHostedPipeline.from_pipelinesd(C(( NNO '/A )2&0J4D"E     rc"0SUR0E$)zGet the identifying parameters.r6)r6rHs r_identifying_params&SelfHostedPipeline._identifying_paramss 4==)  rcg)Nself_hosted_llmr@rTs r _llm_typeSelfHostedPipeline._llm_types rrr run_managerc @UR"SURXS.UD6$)N)rrr r@r4r3)rHrr r[rs r_callSelfHostedPipeline._calls-{{ &&v DJ  rr])Nr)NN)$__name__ __module__ __qualname____firstlineno____doc__r3r__annotations__r4rr5rr6r8rdictr9rrr:boolr model_configrC classmethodintr rQpropertyrrUrYrr^__static_attributes__ __classcell__)rLs@rr1r1Bs<|L#FC+L(+<Hc<<%)NHTN)?!7OJS +E,1#T1L# # J +/    T#Y'         6 WS#X%6  !3!! %):>    tCy!  67      rr1)importlib.utilr#loggingr!typingrrrrrlangchain_core.callbacksr#langchain_core.language_models.llmsr pydanticr langchain_community.llms.utilsr getLoggerr`r)rrrjr/r1r@rrrvs 99=3>   8 $!%    49      (sCC>j j r