Source code for langchain_huggingface.chat_models.huggingface
"""Hugging Face Chat Wrapper."""importjsonfromcollections.abcimportAsyncIterator,Iterator,Mapping,SequencefromdataclassesimportdataclassfromoperatorimportitemgetterfromtypingimportAny,Callable,Literal,Optional,Union,castfromlangchain_core.callbacks.managerimport(AsyncCallbackManagerForLLMRun,CallbackManagerForLLMRun,)fromlangchain_core.language_modelsimportLanguageModelInputfromlangchain_core.language_models.chat_modelsimport(BaseChatModel,agenerate_from_stream,generate_from_stream,)fromlangchain_core.messagesimport(AIMessage,AIMessageChunk,BaseMessage,BaseMessageChunk,ChatMessage,ChatMessageChunk,FunctionMessage,FunctionMessageChunk,HumanMessage,HumanMessageChunk,InvalidToolCall,SystemMessage,SystemMessageChunk,ToolCall,ToolMessage,ToolMessageChunk,)fromlangchain_core.messages.toolimportToolCallChunkfromlangchain_core.messages.toolimporttool_call_chunkascreate_tool_call_chunkfromlangchain_core.output_parsersimportJsonOutputParserfromlangchain_core.output_parsers.openai_toolsimport(JsonOutputKeyToolsParser,make_invalid_tool_call,parse_tool_call,)fromlangchain_core.outputsimport(ChatGeneration,ChatGenerationChunk,ChatResult,LLMResult,)fromlangchain_core.runnablesimportRunnable,RunnableMap,RunnablePassthroughfromlangchain_core.toolsimportBaseToolfromlangchain_core.utils.function_callingimport(convert_to_json_schema,convert_to_openai_tool,)fromlangchain_core.utils.pydanticimportis_basemodel_subclassfrompydanticimportBaseModel,Field,model_validatorfromtyping_extensionsimportSelffrom..llms.huggingface_endpointimportHuggingFaceEndpointfrom..llms.huggingface_pipelineimportHuggingFacePipeline
[docs]@dataclassclassTGI_RESPONSE:"""Response from the TextGenInference API."""choices:list[Any]usage:dict
[docs]@dataclassclassTGI_MESSAGE:"""Message to send to the TextGenInference API."""role:strcontent:strtool_calls:list[dict]
def_lc_tool_call_to_hf_tool_call(tool_call:ToolCall)->dict:return{"type":"function","id":tool_call["id"],"function":{"name":tool_call["name"],"arguments":json.dumps(tool_call["args"]),},}def_lc_invalid_tool_call_to_hf_tool_call(invalid_tool_call:InvalidToolCall,)->dict:return{"type":"function","id":invalid_tool_call["id"],"function":{"name":invalid_tool_call["name"],"arguments":invalid_tool_call["args"],},}def_convert_message_to_dict(message:BaseMessage)->dict:"""Convert a LangChain message to a dictionary. Args: message: The LangChain message. Returns: The dictionary. """message_dict:dict[str,Any]ifisinstance(message,ChatMessage):message_dict={"role":message.role,"content":message.content}elifisinstance(message,HumanMessage):message_dict={"role":"user","content":message.content}elifisinstance(message,AIMessage):message_dict={"role":"assistant","content":message.content}if"function_call"inmessage.additional_kwargs:message_dict["function_call"]=message.additional_kwargs["function_call"]# If function call only, content is None not empty stringifmessage_dict["content"]=="":message_dict["content"]=Noneifmessage.tool_callsormessage.invalid_tool_calls:message_dict["tool_calls"]=[_lc_tool_call_to_hf_tool_call(tc)fortcinmessage.tool_calls]+[_lc_invalid_tool_call_to_hf_tool_call(tc)fortcinmessage.invalid_tool_calls]elif"tool_calls"inmessage.additional_kwargs:message_dict["tool_calls"]=message.additional_kwargs["tool_calls"]# If tool calls only, content is None not empty stringif"tool_calls"inmessage_dictandmessage_dict["content"]=="":message_dict["content"]=Noneelse:passelifisinstance(message,SystemMessage):message_dict={"role":"system","content":message.content}elifisinstance(message,FunctionMessage):message_dict={"role":"function","content":message.content,"name":message.name,}elifisinstance(message,ToolMessage):message_dict={"role":"tool","content":message.content,"tool_call_id":message.tool_call_id,}else:raiseTypeError(f"Got unknown type {message}")if"name"inmessage.additional_kwargs:message_dict["name"]=message.additional_kwargs["name"]returnmessage_dictdef_convert_dict_to_message(_dict:Mapping[str,Any])->BaseMessage:"""Convert a dictionary to a LangChain message. Args: _dict: The dictionary. Returns: The LangChain message. """role=_dict.get("role")ifrole=="user":returnHumanMessage(content=_dict.get("content",""))elifrole=="assistant":content=_dict.get("content","")or""additional_kwargs:dict={}iffunction_call:=_dict.get("function_call"):additional_kwargs["function_call"]=dict(function_call)tool_calls=[]invalid_tool_calls=[]ifraw_tool_calls:=_dict.get("tool_calls"):additional_kwargs["tool_calls"]=raw_tool_callsforraw_tool_callinraw_tool_calls:try:tool_calls.append(parse_tool_call(raw_tool_call,return_id=True))exceptExceptionase:invalid_tool_calls.append(dict(make_invalid_tool_call(raw_tool_call,str(e))))returnAIMessage(content=content,additional_kwargs=additional_kwargs,tool_calls=tool_calls,invalid_tool_calls=invalid_tool_calls,)elifrole=="system":returnSystemMessage(content=_dict.get("content",""))elifrole=="function":returnFunctionMessage(content=_dict.get("content",""),name=_dict.get("name",""))elifrole=="tool":additional_kwargs={}if"name"in_dict:additional_kwargs["name"]=_dict["name"]returnToolMessage(content=_dict.get("content",""),tool_call_id=_dict.get("tool_call_id",""),additional_kwargs=additional_kwargs,)else:returnChatMessage(content=_dict.get("content",""),role=roleor"")def_is_huggingface_hub(llm:Any)->bool:try:fromlangchain_community.llms.huggingface_hubimport(HuggingFaceHub,# type: ignore[import-not-found])returnisinstance(llm,HuggingFaceHub)exceptImportError:# if no langchain community, it is not a HuggingFaceHubreturnFalsedef_convert_chunk_to_message_chunk(chunk:Mapping[str,Any],default_class:type[BaseMessageChunk])->BaseMessageChunk:choice=chunk["choices"][0]_dict=choice["delta"]role=cast(str,_dict.get("role"))content=cast(str,_dict.get("content")or"")additional_kwargs:dict={}tool_call_chunks:list[ToolCallChunk]=[]if_dict.get("function_call"):function_call=dict(_dict["function_call"])if"name"infunction_callandfunction_call["name"]isNone:function_call["name"]=""additional_kwargs["function_call"]=function_callifraw_tool_calls:=_dict.get("tool_calls"):additional_kwargs["tool_calls"]=raw_tool_callsforrtcinraw_tool_calls:try:tool_call_chunks.append(create_tool_call_chunk(name=rtc["function"].get("name"),args=rtc["function"].get("arguments"),id=rtc.get("id"),index=rtc.get("index"),))exceptKeyError:passifrole=="user"ordefault_class==HumanMessageChunk:returnHumanMessageChunk(content=content)elifrole=="assistant"ordefault_class==AIMessageChunk:ifusage:=chunk.get("usage"):input_tokens=usage.get("prompt_tokens",0)output_tokens=usage.get("completion_tokens",0)usage_metadata={"input_tokens":input_tokens,"output_tokens":output_tokens,"total_tokens":usage.get("total_tokens",input_tokens+output_tokens),}else:usage_metadata=NonereturnAIMessageChunk(content=content,additional_kwargs=additional_kwargs,tool_call_chunks=tool_call_chunks,usage_metadata=usage_metadata,# type: ignore[arg-type])elifrole=="system"ordefault_class==SystemMessageChunk:returnSystemMessageChunk(content=content)elifrole=="function"ordefault_class==FunctionMessageChunk:returnFunctionMessageChunk(content=content,name=_dict["name"])elifrole=="tool"ordefault_class==ToolMessageChunk:returnToolMessageChunk(content=content,tool_call_id=_dict["tool_call_id"])elifroleordefault_class==ChatMessageChunk:returnChatMessageChunk(content=content,role=role)else:returndefault_class(content=content)# type: ignoredef_is_huggingface_textgen_inference(llm:Any)->bool:try:fromlangchain_community.llms.huggingface_text_gen_inferenceimport(HuggingFaceTextGenInference,# type: ignore[import-not-found])returnisinstance(llm,HuggingFaceTextGenInference)exceptImportError:# if no langchain community, it is not a HuggingFaceTextGenInferencereturnFalsedef_is_huggingface_endpoint(llm:Any)->bool:returnisinstance(llm,HuggingFaceEndpoint)def_is_huggingface_pipeline(llm:Any)->bool:returnisinstance(llm,HuggingFacePipeline)
[docs]classChatHuggingFace(BaseChatModel):"""Hugging Face LLM's as ChatModels. Works with `HuggingFaceTextGenInference`, `HuggingFaceEndpoint`, `HuggingFaceHub`, and `HuggingFacePipeline` LLMs. Upon instantiating this class, the model_id is resolved from the url provided to the LLM, and the appropriate tokenizer is loaded from the HuggingFace Hub. Setup: Install ``langchain-huggingface`` and ensure your Hugging Face token is saved. .. code-block:: bash pip install langchain-huggingface .. code-block:: python from huggingface_hub import login login() # You will be prompted for your HF key, which will then be saved locally Key init args — completion params: llm: `HuggingFaceTextGenInference`, `HuggingFaceEndpoint`, `HuggingFaceHub`, or 'HuggingFacePipeline' LLM to be used. Key init args — client params: custom_get_token_ids: Optional[Callable[[str], list[int]]] Optional encoder to use for counting tokens. metadata: Optional[dict[str, Any]] Metadata to add to the run trace. tags: Optional[list[str]] Tags to add to the run trace. tokenizer: Any verbose: bool Whether to print out response text. See full list of supported init args and their descriptions in the params section. Instantiate: .. code-block:: python from langchain_huggingface import HuggingFaceEndpoint, ChatHuggingFace llm = HuggingFaceEndpoint( repo_id="microsoft/Phi-3-mini-4k-instruct", task="text-generation", max_new_tokens=512, do_sample=False, repetition_penalty=1.03, ) chat = ChatHuggingFace(llm=llm, verbose=True) Invoke: .. code-block:: python messages = [ ("system", "You are a helpful translator. Translate the user sentence to French."), ("human", "I love programming."), ] chat(...).invoke(messages) .. code-block:: python AIMessage(content='Je ai une passion pour le programme.\n\nIn French, we use "ai" for masculine subjects and "a" for feminine subjects. Since "programming" is gender-neutral in English, we will go with the masculine "programme".\n\nConfirmation: "J\'aime le programme." is more commonly used. The sentence above is technically accurate, but less commonly used in spoken French as "ai" is used less frequently in everyday speech.', response_metadata={'token_usage': ChatCompletionOutputUsage (completion_tokens=100, prompt_tokens=55, total_tokens=155), 'model': '', 'finish_reason': 'length'}, id='run-874c24b7-0272-4c99-b259-5d6d7facbc56-0') Stream: .. code-block:: python for chunk in chat.stream(messages): print(chunk) .. code-block:: python content='Je ai une passion pour le programme.\n\nIn French, we use "ai" for masculine subjects and "a" for feminine subjects. Since "programming" is gender-neutral in English, we will go with the masculine "programme".\n\nConfirmation: "J\'aime le programme." is more commonly used. The sentence above is technically accurate, but less commonly used in spoken French as "ai" is used less frequently in everyday speech.' response_metadata={'token_usage': ChatCompletionOutputUsage (completion_tokens=100, prompt_tokens=55, total_tokens=155), 'model': '', 'finish_reason': 'length'} id='run-7d7b1967-9612-4f9a-911a-b2b5ca85046a-0' Async: .. code-block:: python await chat.ainvoke(messages) .. code-block:: python AIMessage(content='Je déaime le programming.\n\nLittérale : Je (j\'aime) déaime (le) programming.\n\nNote: "Programming" in French is "programmation". But here, I used "programming" instead of "programmation" because the user said "I love programming" instead of "I love programming (in French)", which would be "J\'aime la programmation". By translating the sentence literally, I preserved the original meaning of the user\'s sentence.', id='run-fd850318-e299-4735-b4c6-3496dc930b1d-0') Tool calling: .. code-block:: python from pydantic import BaseModel, Field class GetWeather(BaseModel): '''Get the current weather in a given location''' location: str = Field(..., description="The city and state, e.g. San Francisco, CA") class GetPopulation(BaseModel): '''Get the current population in a given location''' location: str = Field(..., description="The city and state, e.g. San Francisco, CA") chat_with_tools = chat.bind_tools([GetWeather, GetPopulation]) ai_msg = chat_with_tools.invoke("Which city is hotter today and which is bigger: LA or NY?") ai_msg.tool_calls .. code-block:: python [{'name': 'GetPopulation', 'args': {'location': 'Los Angeles, CA'}, 'id': '0'}] Response metadata .. code-block:: python ai_msg = chat.invoke(messages) ai_msg.response_metadata .. code-block:: python {'token_usage': ChatCompletionOutputUsage(completion_tokens=100, prompt_tokens=8, total_tokens=108), 'model': '', 'finish_reason': 'length'} """# noqa: E501llm:Any"""LLM, must be of type HuggingFaceTextGenInference, HuggingFaceEndpoint, HuggingFaceHub, or HuggingFacePipeline."""tokenizer:Any=None"""Tokenizer for the model. Only used for HuggingFacePipeline."""model_id:Optional[str]=None"""Model ID for the model. Only used for HuggingFaceEndpoint."""temperature:Optional[float]=None"""What sampling temperature to use."""stop:Optional[Union[str,list[str]]]=Field(default=None,alias="stop_sequences")"""Default stop sequences."""presence_penalty:Optional[float]=None"""Penalizes repeated tokens."""frequency_penalty:Optional[float]=None"""Penalizes repeated tokens according to frequency."""seed:Optional[int]=None"""Seed for generation"""logprobs:Optional[bool]=None"""Whether to return logprobs."""top_logprobs:Optional[int]=None"""Number of most likely tokens to return at each token position, each with an associated log probability. `logprobs` must be set to true if this parameter is used."""logit_bias:Optional[dict[int,int]]=None"""Modify the likelihood of specified tokens appearing in the completion."""streaming:bool=False"""Whether to stream the results or not."""n:Optional[int]=None"""Number of chat completions to generate for each prompt."""top_p:Optional[float]=None"""Total probability mass of tokens to consider at each step."""max_tokens:Optional[int]=None"""Maximum number of tokens to generate."""model_kwargs:dict[str,Any]=Field(default_factory=dict)"""Holds any model parameters valid for `create` call not explicitly specified."""def__init__(self,**kwargs:Any):super().__init__(**kwargs)self._resolve_model_id()@model_validator(mode="after")defvalidate_llm(self)->Self:if(not_is_huggingface_hub(self.llm)andnot_is_huggingface_textgen_inference(self.llm)andnot_is_huggingface_endpoint(self.llm)andnot_is_huggingface_pipeline(self.llm)):raiseTypeError("Expected llm to be one of HuggingFaceTextGenInference, ""HuggingFaceEndpoint, HuggingFaceHub, HuggingFacePipeline "f"received {type(self.llm)}")returnselfdef_create_chat_result(self,response:dict)->ChatResult:generations=[]token_usage=response.get("usage",{})forresinresponse["choices"]:message=_convert_dict_to_message(res["message"])iftoken_usageandisinstance(message,AIMessage):message.usage_metadata={"input_tokens":token_usage.get("prompt_tokens",0),"output_tokens":token_usage.get("completion_tokens",0),"total_tokens":token_usage.get("total_tokens",0),}generation_info=dict(finish_reason=res.get("finish_reason"))if"logprobs"inres:generation_info["logprobs"]=res["logprobs"]gen=ChatGeneration(message=message,generation_info=generation_info,)generations.append(gen)llm_output={"token_usage":token_usage,"model_name":self.model_id,"system_fingerprint":response.get("system_fingerprint",""),}returnChatResult(generations=generations,llm_output=llm_output)def_generate(self,messages:list[BaseMessage],stop:Optional[list[str]]=None,run_manager:Optional[CallbackManagerForLLMRun]=None,stream:Optional[bool]=None,**kwargs:Any,)->ChatResult:should_stream=streamifstreamisnotNoneelseself.streamingif_is_huggingface_textgen_inference(self.llm):message_dicts,params=self._create_message_dicts(messages,stop)answer=self.llm.client.chat(messages=message_dicts,**kwargs)returnself._create_chat_result(answer)elif_is_huggingface_endpoint(self.llm):ifshould_stream:stream_iter=self._stream(messages,stop=stop,run_manager=run_manager,**kwargs)returngenerate_from_stream(stream_iter)message_dicts,params=self._create_message_dicts(messages,stop)params={"stop":stop,**params,**({"stream":stream}ifstreamisnotNoneelse{}),**kwargs,}answer=self.llm.client.chat_completion(messages=message_dicts,**params)returnself._create_chat_result(answer)else:llm_input=self._to_chat_prompt(messages)ifshould_stream:stream_iter=self.llm._stream(llm_input,stop=stop,run_manager=run_manager,**kwargs)returngenerate_from_stream(stream_iter)llm_result=self.llm._generate(prompts=[llm_input],stop=stop,run_manager=run_manager,**kwargs)returnself._to_chat_result(llm_result)asyncdef_agenerate(self,messages:list[BaseMessage],stop:Optional[list[str]]=None,run_manager:Optional[AsyncCallbackManagerForLLMRun]=None,stream:Optional[bool]=None,**kwargs:Any,)->ChatResult:if_is_huggingface_textgen_inference(self.llm):message_dicts,params=self._create_message_dicts(messages,stop)answer=awaitself.llm.async_client.chat(messages=message_dicts,**kwargs)returnself._create_chat_result(answer)elif_is_huggingface_endpoint(self.llm):should_stream=streamifstreamisnotNoneelseself.streamingifshould_stream:stream_iter=self._astream(messages,stop=stop,run_manager=run_manager,**kwargs)returnawaitagenerate_from_stream(stream_iter)message_dicts,params=self._create_message_dicts(messages,stop)params={**params,**({"stream":stream}ifstreamisnotNoneelse{}),**kwargs,}answer=awaitself.llm.async_client.chat_completion(messages=message_dicts,**params)returnself._create_chat_result(answer)elif_is_huggingface_pipeline(self.llm):raiseNotImplementedError("async generation is not supported with HuggingFacePipeline")else:llm_input=self._to_chat_prompt(messages)llm_result=awaitself.llm._agenerate(prompts=[llm_input],stop=stop,run_manager=run_manager,**kwargs)returnself._to_chat_result(llm_result)def_stream(self,messages:list[BaseMessage],stop:Optional[list[str]]=None,run_manager:Optional[CallbackManagerForLLMRun]=None,**kwargs:Any,)->Iterator[ChatGenerationChunk]:if_is_huggingface_endpoint(self.llm):message_dicts,params=self._create_message_dicts(messages,stop)params={**params,**kwargs,"stream":True}default_chunk_class:type[BaseMessageChunk]=AIMessageChunkforchunkinself.llm.client.chat_completion(messages=message_dicts,**params):iflen(chunk["choices"])==0:continuechoice=chunk["choices"][0]message_chunk=_convert_chunk_to_message_chunk(chunk,default_chunk_class)generation_info={}iffinish_reason:=choice.get("finish_reason"):generation_info["finish_reason"]=finish_reasongeneration_info["model_name"]=self.model_idlogprobs=choice.get("logprobs")iflogprobs:generation_info["logprobs"]=logprobsdefault_chunk_class=message_chunk.__class__generation_chunk=ChatGenerationChunk(message=message_chunk,generation_info=generation_infoorNone)ifrun_manager:run_manager.on_llm_new_token(generation_chunk.text,chunk=generation_chunk,logprobs=logprobs)yieldgeneration_chunkelse:llm_input=self._to_chat_prompt(messages)stream_iter=self.llm._stream(llm_input,stop=stop,run_manager=run_manager,**kwargs)forchunkinstream_iter:# chunk is a GenerationChunkchat_chunk=ChatGenerationChunk(message=AIMessageChunk(content=chunk.text),generation_info=chunk.generation_info,)yieldchat_chunkasyncdef_astream(self,messages:list[BaseMessage],stop:Optional[list[str]]=None,run_manager:Optional[AsyncCallbackManagerForLLMRun]=None,**kwargs:Any,)->AsyncIterator[ChatGenerationChunk]:message_dicts,params=self._create_message_dicts(messages,stop)params={**params,**kwargs,"stream":True}default_chunk_class:type[BaseMessageChunk]=AIMessageChunkasyncforchunkinawaitself.llm.async_client.chat_completion(messages=message_dicts,**params):iflen(chunk["choices"])==0:continuechoice=chunk["choices"][0]message_chunk=_convert_chunk_to_message_chunk(chunk,default_chunk_class)generation_info={}iffinish_reason:=choice.get("finish_reason"):generation_info["finish_reason"]=finish_reasongeneration_info["model_name"]=self.model_idlogprobs=choice.get("logprobs")iflogprobs:generation_info["logprobs"]=logprobsdefault_chunk_class=message_chunk.__class__generation_chunk=ChatGenerationChunk(message=message_chunk,generation_info=generation_infoorNone)ifrun_manager:awaitrun_manager.on_llm_new_token(token=generation_chunk.text,chunk=generation_chunk,logprobs=logprobs,)yieldgeneration_chunkdef_to_chat_prompt(self,messages:list[BaseMessage],)->str:"""Convert a list of messages into a prompt format expected by wrapped LLM."""ifnotmessages:raiseValueError("At least one HumanMessage must be provided!")ifnotisinstance(messages[-1],HumanMessage):raiseValueError("Last message must be a HumanMessage!")messages_dicts=[self._to_chatml_format(m)forminmessages]returnself.tokenizer.apply_chat_template(messages_dicts,tokenize=False,add_generation_prompt=True)def_to_chatml_format(self,message:BaseMessage)->dict:"""Convert LangChain message to ChatML format."""ifisinstance(message,SystemMessage):role="system"elifisinstance(message,AIMessage):role="assistant"elifisinstance(message,HumanMessage):role="user"else:raiseValueError(f"Unknown message type: {type(message)}")return{"role":role,"content":message.content}@staticmethoddef_to_chat_result(llm_result:LLMResult)->ChatResult:chat_generations=[]forginllm_result.generations[0]:chat_generation=ChatGeneration(message=AIMessage(content=g.text),generation_info=g.generation_info)chat_generations.append(chat_generation)returnChatResult(generations=chat_generations,llm_output=llm_result.llm_output)def_resolve_model_id(self)->None:"""Resolve the model_id from the LLM's inference_server_url"""fromhuggingface_hubimportlist_inference_endpoints# type: ignore[import]if_is_huggingface_hub(self.llm)or(hasattr(self.llm,"repo_id")andself.llm.repo_id):self.model_id=self.llm.repo_idreturnelif_is_huggingface_textgen_inference(self.llm):endpoint_url:Optional[str]=self.llm.inference_server_urlelif_is_huggingface_pipeline(self.llm):fromtransformersimportAutoTokenizer# type: ignore[import]self.tokenizer=(AutoTokenizer.from_pretrained(self.model_id)ifself.tokenizerisNoneelseself.tokenizer)self.model_id=self.llm.model_idreturnelif_is_huggingface_endpoint(self.llm):self.model_id=self.llm.repo_idorself.llm.modelreturnelse:endpoint_url=self.llm.endpoint_urlavailable_endpoints=list_inference_endpoints("*")forendpointinavailable_endpoints:ifendpoint.url==endpoint_url:self.model_id=endpoint.repositoryifnotself.model_id:raiseValueError("Failed to resolve model_id:"f"Could not find model id for inference server: {endpoint_url}""Make sure that your Hugging Face token has access to the endpoint.")
[docs]defbind_tools(self,tools:Sequence[Union[dict[str,Any],type,Callable,BaseTool]],*,tool_choice:Optional[Union[dict,str,Literal["auto","none","required"],bool]]=None,**kwargs:Any,)->Runnable[LanguageModelInput,BaseMessage]:"""Bind tool-like objects to this chat model. Assumes model is compatible with OpenAI tool-calling API. Args: tools: A list of tool definitions to bind to this chat model. Supports any tool definition handled by :meth:`langchain_core.utils.function_calling.convert_to_openai_tool`. tool_choice: Which tool to require the model to call. Must be the name of the single provided function or "auto" to automatically determine which function to call (if any), or a dict of the form: {"type": "function", "function": {"name": <<tool_name>>}}. **kwargs: Any additional parameters to pass to the :class:`~langchain.runnable.Runnable` constructor. """formatted_tools=[convert_to_openai_tool(tool)fortoolintools]iftool_choiceisnotNoneandtool_choice:iflen(formatted_tools)!=1:raiseValueError("When specifying `tool_choice`, you must provide exactly one "f"tool. Received {len(formatted_tools)} tools.")ifisinstance(tool_choice,str):iftool_choicenotin("auto","none","required"):tool_choice={"type":"function","function":{"name":tool_choice},}elifisinstance(tool_choice,bool):tool_choice=formatted_tools[0]elifisinstance(tool_choice,dict):if(formatted_tools[0]["function"]["name"]!=tool_choice["function"]["name"]):raiseValueError(f"Tool choice {tool_choice} was specified, but the only "f"provided tool was {formatted_tools[0]['function']['name']}.")else:raiseValueError(f"Unrecognized tool_choice type. Expected str, bool or dict. "f"Received: {tool_choice}")kwargs["tool_choice"]=tool_choicereturnsuper().bind(tools=formatted_tools,**kwargs)
[docs]defwith_structured_output(self,schema:Optional[Union[dict,type[BaseModel]]]=None,*,method:Literal["function_calling","json_mode","json_schema"]="function_calling",include_raw:bool=False,**kwargs:Any,)->Runnable[LanguageModelInput,Union[dict,BaseModel]]:"""Model wrapper that returns outputs formatted to match the given schema. Args: schema: The output schema. Can be passed in as: - an OpenAI function/tool schema, - a JSON Schema, - a typedDict class (support added in 0.1.7), Pydantic class is currently supported. method: The method for steering model generation, one of: - "function_calling": uses tool-calling features. - "json_schema": uses dedicated structured output features. - "json_mode": uses JSON mode. include_raw: If False then only the parsed structured output is returned. If an error occurs during model output parsing it will be raised. If True then both the raw model response (a BaseMessage) and the parsed model response will be returned. If an error occurs during output parsing it will be caught and returned as well. The final output is always a dict with keys "raw", "parsed", and "parsing_error". Returns: A Runnable that takes same inputs as a :class:`langchain_core.language_models.chat.BaseChatModel`. If ``include_raw`` is False and ``schema`` is a Pydantic class, Runnable outputs an instance of ``schema`` (i.e., a Pydantic object). Otherwise, if ``include_raw`` is False then Runnable outputs a dict. If ``include_raw`` is True, then Runnable outputs a dict with keys: - ``"raw"``: BaseMessage - ``"parsed"``: None if there was a parsing error, otherwise the type depends on the ``schema`` as described above. - ``"parsing_error"``: Optional[BaseException] """# noqa: E501_=kwargs.pop("strict",None)ifkwargs:raiseValueError(f"Received unsupported arguments {kwargs}")is_pydantic_schema=isinstance(schema,type)andis_basemodel_subclass(schema)ifmethod=="function_calling":ifschemaisNone:raiseValueError("schema must be specified when method is 'function_calling'. ""Received None.")formatted_tool=convert_to_openai_tool(schema)tool_name=formatted_tool["function"]["name"]llm=self.bind_tools([schema],tool_choice=tool_name,ls_structured_output_format={"kwargs":{"method":"function_calling"},"schema":formatted_tool,},)ifis_pydantic_schema:raiseNotImplementedError("Pydantic schema is not supported for function calling")else:output_parser:Union[JsonOutputKeyToolsParser,JsonOutputParser]=(JsonOutputKeyToolsParser(key_name=tool_name,first_tool_only=True))elifmethod=="json_schema":ifschemaisNone:raiseValueError("schema must be specified when method is 'json_schema'. ""Received None.")formatted_schema=convert_to_json_schema(schema)llm=self.bind(response_format={"type":"json_object","schema":formatted_schema},ls_structured_output_format={"kwargs":{"method":"json_schema"},"schema":schema,},)output_parser:Union[# type: ignore[no-redef]JsonOutputKeyToolsParser,JsonOutputParser]=JsonOutputParser()# type: ignore[arg-type]elifmethod=="json_mode":llm=self.bind(response_format={"type":"json_object"},ls_structured_output_format={"kwargs":{"method":"json_mode"},"schema":schema,},)output_parser:Union[# type: ignore[no-redef]JsonOutputKeyToolsParser,JsonOutputParser]=JsonOutputParser()# type: ignore[arg-type]else:raiseValueError(f"Unrecognized method argument. Expected one of 'function_calling' or "f"'json_mode'. Received: '{method}'")ifinclude_raw:parser_assign=RunnablePassthrough.assign(parsed=itemgetter("raw")|output_parser,parsing_error=lambda_:None)parser_none=RunnablePassthrough.assign(parsed=lambda_:None)parser_with_fallback=parser_assign.with_fallbacks([parser_none],exception_key="parsing_error")returnRunnableMap(raw=llm)|parser_with_fallbackelse:returnllm|output_parser
def_create_message_dicts(self,messages:list[BaseMessage],stop:Optional[list[str]])->tuple[list[dict[str,Any]],dict[str,Any]]:params=self._default_paramsifstopisnotNone:params["stop"]=stopmessage_dicts=[_convert_message_to_dict(m)forminmessages]returnmessage_dicts,params@propertydef_default_params(self)->dict[str,Any]:"""Get the default parameters for calling Hugging Face Inference Providers API."""params={"model":self.model_id,"stream":self.streaming,"n":self.n,"temperature":self.temperature,"stop":self.stop,**(self.model_kwargsifself.model_kwargselse{}),}ifself.max_tokensisnotNone:params["max_tokens"]=self.max_tokensreturnparams@propertydef_llm_type(self)->str:return"huggingface-chat-wrapper"