- Notifications
You must be signed in to change notification settings - Fork40
Python SDK Migration Guide
A guide for developers migrating to the Fern-generated Python SDK (version0.7.0 and above).
Hume’s newest Python SDK refactors the core client architecture, separating functionality into distinct modules for specific APIs (e.g., theExpression Measurement API and theEmpathic Voice Interface API).
Version0.7.0 introduces the following features:
- Explicit types
- Better support for asynchronous operations
- More granular client configuration
- Continued support for legacy SDK implementations
- Support for Python version
3.12with Expression Measurement API namespace methods
This guide will help you adapt your code to the new SDK structure with practical examples and explanations of the key differences.
Below is a matrix showing the compatibility of the Hume Python SDK across various Python versions and operating systems.
| Python Version | Operating System | |
|---|---|---|
| Empathic Voice Interface | 3.9,3.10,3.11 | macOS, Linux |
| Expression Measurement | 3.9,3.10,3.11,3.12 | macOS, Linux, Windows |
For the Empathic Voice Interface, Python versions3.9 through3.11 are supported on macOS and Linux.
For Expression Measurement, Python versions3.9 through3.12 are supported on macOS, Linux, and Windows.
The legacy SDK is entirely contained within the new SDK’ssrc/hume/legacy folder in order to ensure smooth transition to the new features. To preserve your code’s current functionality, follow these steps:
- Run
pip install “hume[legacy]"to install the legacy package extra.- If you are using EVI’s microphone utilities, run
pip install “hume[microphone]”to install the microphone extra.
- If you are using EVI’s microphone utilities, run
- Change your import statements to
from hume.legacyinstead offrom hume.
fromhume.legacyimportHumeVoiceClient,VoiceConfigclient=HumeVoiceClient("<YOUR_API_KEY>")config=client.empathic_voice.configs.get_config_version(id="id",version=1 )
Instead of usingHumeBatchClient,HumeStreamClient, orHumeVoiceClient, now useAsyncHumeClient - the new asynchronous base client.
This client is authenticated with your Hume API key and provides access to the Expression Measurement API and Empathic Voice Interface API as namespaces. If you're not using async, the synchronousHumeClient is available, but we recommend defaulting toAsyncHumeClient for most use cases.
Each API is namespaced accordingly:
fromhume.clientimportAsyncHumeClient# base synchronous clientclient=AsyncHumeClient(api_key=<HUME_API_KEY>)# Expression Measurement (Batch)client.expression_measurement.batch# Expression Measurement (Streaming)client.expression_measurement.streaming# Empathic Voice Interfaceclient.empathic_voice.
Importantly, invoking asynchronous functionality (e.g., instantiating an EVI WebSocket connection) when using a synchronous client (i.e.,HumeClient) is disallowed behavior and causes an error. On the other hand, invoking synchronous behavior from an asynchronous client is supported, however each method must be awaited.
fromhume.clientimportHumeClient,AsyncHumeClient# INVALID: using a synchronous client for asynchronous behaviorclient=HumeClient(api_key=<HUME_API_KEY>)# Using the asynchronous connect method with a sync client will cause an errorasyncwithclient.empathic_voice.chat.connect()assocket:# ...# VALID: using an asynchronous client for asynchronous behaviorasync_client=AsyncHumeClient(api_key=<HUME_API_KEY>)# Using the async connect method with an async client will work properlyasyncwithasync_client.empathic_voice.chat.connect()assocket:# ...# VALID: using an asynchronous client for synchronous behaviorasync_client=AsyncHumeClient(api_key=<HUME_API_KEY>)# Using the configs.list_configs() method with an async clientprint(awaitclient.empathic_voice.configs.list_configs())
First, identify what operations you would like to perform.
- For tasks such as creating a config, listing the tools you have available, and more, we recommend using theHume Portal because of its comprehensive user interface.
- For chatting with EVI (i.e., accessing the
chatendpoint), it is required to use the asynchronous Hume client. - If you need to interact with configurations, tools, or other items programmatically, it is recommended to use the asynchronous Hume client - but possible to use the synchronous client if needed.
Then, authenticate the client and proceed with your desired functionality.
The EVI WebSocket connection is now configurable using an explicit type:ChatConnectOptions. This object must be passed into the method used to initialize the connection.
fromhume.clientimportHumeClient# authenticate the synchronous clientclient=HumeClient(api_key=<HUME_API_KEY>)# list your configsclient.empathic_voice.configs.list_configs()
It is now possible to fully manage the WebSocket events with your EVI integration, meaning you can define custom behavior when the WebSocket is opened, closed, receives a message, or receives an error. Use the new asynchronous client’sconnect_with_callbacks function to do so, and reference theSubscribeEvent message type within youron_message callback function.
fromhume.clientimportAsyncHumeClientfromhume.empathic_voice.chat.socket_clientimportChatConnectOptionsasyncdefmain()->None:# Initialize the asynchronous client, authenticating with your API keyclient=AsyncHumeClient(api_key=<HUME_API_KEY>)# Define options for the WebSocket connection, such as an EVI config id and a secret key for token authenticationoptions=ChatConnectOptions(config_id=<HUME_CONFIG_ID>,secret_key=<HUME_SECRET_KEY>)# Open the WebSocket connection with the configuration options and the interface's handlersasyncwithclient.empathic_voice.chat.connect_with_callbacks(options=options,on_open=<customon_openfunction>,on_message=<customon_messagefunction>,on_close=<customon_closefunction>,on_error=<customon_errorfunction> )assocket:# ...if__name__=="__main__":asyncio.run(main())
asyncdefon_message(message:SubscribeEvent):"""Callback function to handle a WebSocket message event. Args: data (SubscribeEvent): This represents any type of message that is received through the EVI WebSocket, formatted in JSON. See the full list of messages in the API Reference [here](https://dev.hume.ai/reference/empathic-voice-interface-evi/chat/chat#receive). """# Create an empty dictionary to store expression inference scoresscores= {}ifmessage.type=="chat_metadata":message_type=message.type.upper()chat_id=message.chat_idchat_group_id=message.chat_group_idtext=f"<{message_type}> Chat ID:{chat_id}, Chat Group ID:{chat_group_id}"elifmessage.typein ["user_message","assistant_message"]:role=message.message.role.upper()message_text=message.message.contenttext=f"{role}:{message_text}"ifmessage.from_textisFalse:scores=dict(message.models.prosody.scores)elifmessage.type=="audio_output":message_str:str=message.datamessage_bytes=base64.b64decode(message_str.encode("utf-8"))awaitself.byte_strs.put(message_bytes)returnelifmessage.type=="error":error_message:str=message.messageerror_code:str=message.coderaiseApiError(f"Error ({error_code}):{error_message}")# ApiError is also an imported typeelse:message_type=message.type.upper()text=f"<{message_type}>"print(text)
fromhumeimportHumeVoiceClient,MicrophoneInterfaceimportasyncioasyncdefmain()->None:# Connect and authenticate with Humeclient=HumeVoiceClient(<HUME_API_KEY>)# Start streaming EVI over your device's microphone and speakersasyncwithclient.connect()assocket:awaitMicrophoneInterface.start(socket)if__name__=="__main__":asyncio.run(main())
Instantiate the asynchronous client, configure the job with aModels object, and submit your media URLs for processing. Once submitted and the job is awaited to completion, predictions may be retrieved based on the job ID.
- The
await_complete()method on a job has been removed; developers will need to implement a mechanism such as polling the job’s status to await the completion of the job. - The
download_predictions()method on a job has also been removed; developers will need to implement an HTTP call to the API, parse the results, and export them to a file.
Prior to the update, when you started a job and passed in the job configuration, it would be the case that thestart_inference_job would accept the model configs as an array. Now, this is all contained within a typed models object.
Starting an inference job now involves defining configuration options using explicit types for each model. For example, a Face object corresponds to the model’s configuration options. Configurations are passed into a Models object, which in turn is passed into the start_inference_job method. Similar strict typing exists with other batch methods.
fromhumeimportAsyncHumeClientfromhume.expression_measurement.batchimportFace,Modelsasyncdefmain():# Initialize an authenticated clientclient=AsyncHumeClient(api_key=<YOUR_API_KEY>)# Define the URL(s) of the files you would like to analyzejob_urls= ["https://hume-tutorials.s3.amazonaws.com/faces.zip"]# Create configurations for each model you would like to use (blank = default)face_config=Face()# Create a Models objectmodels_chosen=Models(face=face_config)# Start an inference job and print the job_idjob_id=awaitclient.expression_measurement.batch.start_inference_job(urls=job_urls,models=models_chosen )# Await the completion of the inference jobawaitpoll_for_completion(client,job_id,timeout=120)# After the job is over, access its predictionsjob_predictions=awaitclient.expression_measurement.batch.get_job_predictions(id=job_id )if__name__=="__main__":asyncio.run(main())
fromhumeimportAsyncHumeClientfromhume.expression_measurement.batchimportFace,Modelsfromhume.expression_measurement.batch.typesimportInferenceBaseRequestasyncdefmain():# Initialize an authenticated clientclient=AsyncHumeClient(api_key=HUME_API_KEY)# Define the filepath(s) of the file(s) you would like to analyzelocal_filepaths= [open("faces.zip",mode="rb")]# Create configurations for each model you would like to use (blank = default)face_config=Face()# Create a Models objectmodels_chosen=Models(face=face_config)# Create a stringified object containing the configurationstringified_configs=InferenceBaseRequest(models=models_chosen)# Start an inference job and print the job_idjob_id=awaitclient.expression_measurement.batch.start_inference_job_from_local_file(json=stringified_configs,file=local_filepaths)# Await the completion of the inference jobawaitpoll_for_completion(client,job_id,timeout=120)# After the job is over, access its predictionsjob_predictions=awaitclient.expression_measurement.batch.get_job_predictions(id=job_id )if__name__=="__main__":asyncio.run(main())
Below is an example implementation of helper methods which incorporate polling the job’s status for completion with exponential backoff.
asyncdefpoll_for_completion(client:AsyncHumeClient,job_id,timeout=120):""" Polls for the completion of a job with a specified timeout (in seconds). Uses asyncio.wait_for to enforce a maximum waiting time. """try:# Wait for the job to complete or until the timeout is reachedawaitasyncio.wait_for(poll_until_complete(client,job_id),timeout=timeout)exceptasyncio.TimeoutError:# Notify if the polling operation has timed outprint(f"Polling timed out after{timeout} seconds.")asyncdefpoll_until_complete(client:AsyncHumeClient,job_id):""" Continuously polls the job status until it is completed, failed, or an unexpected status is encountered. Implements exponential backoff to reduce the frequency of requests over time. """delay=1# Start with a 1-second delaywhileTrue:# Wait for the specified delay before making the next status checkawaitasyncio.sleep(delay)# Retrieve the current job detailsjob_details=awaitclient.expression_measurement.batch.get_job_details(job_id)status=job_details.state.statusifstatus=="COMPLETED":# Job has completed successfullyprint("\nJob completed successfully:")breakelifstatus=="FAILED":# Job has failedprint("\nJob failed:")break# Increase the delay exponentially, maxing out at 16 secondsdelay=min(delay*2,16)
The SDK may be used to download the job’s artifacts.
withopen("artifacts.zip","wb")asf:asyncfornew_bytesinclient.expression_measurement.batch.get_job_artifacts(job_id):f.write(new_bytes)
The API must be called directly to download the job’s predictions.
If using the code below, ensure you replace <YOUR_JOB_ID> and <YOUR_API_KEY> below with the respective correct values.
importrequestsimportjson# Define the URL and headersurl="https://api.hume.ai/v0/batch/jobs/<YOUR_JOB_ID>/predictions"headers= {"X-Hume-Api-Key":"<YOUR_API_KEY>" }# Make the GET requestresponse=requests.get(url,headers=headers)# Check if the request was successfulifresponse.status_code==200:# Parse the JSON responsedata=response.json()# Write the JSON data to a filewithopen("predictions.json","w")asfile:json.dump(data,file,indent=2)print("Response has been written to 'predictions.json'.")else:print(f"Failed to fetch data. Status code:{response.status_code}")print(response.text)
fromhumeimportHumeBatchClientfromhume.models.configimportFaceConfigfromhume.models.configimportProsodyConfigclient=HumeBatchClient(<HUME_API_KEY>)urls= ["https://hume-tutorials.s3.amazonaws.com/faces.zip"]face_config=FaceConfig()prosody_config=ProsodyConfig()job=client.submit_job(urls, [face_config,prosody_config])print(job)print("Running...")result=job.await_complete()job_predictions=client.get_job_predictions(job_id=job.id)
First, retrieve the samples you will use. Then, instantiate the asynchronous client and configure the WebSocket with aConfig object containing the model(s) you would like to use. After you connect to the WebSocket, predictions may be retrieved.
Connecting to the WebSocket now uses the explicit typeStreamConnectOptions. These options accept theConfig object, which contains the configurations for the expression measurement models you wish to use. These configurations are unique to each model and need importing as well, such as withStreamLanguage.
importasynciofromhumeimportAsyncHumeClientfromhume.expression_measurement.streamimportConfigfromhume.expression_measurement.stream.socket_clientimportStreamConnectOptionsfromhume.expression_measurement.stream.typesimportStreamLanguagesamples= ["Mary had a little lamb,","Its fleece was white as snow.""Everywhere the child went,""The little lamb was sure to go." ]asyncdefmain():client=AsyncHumeClient(api_key="<YOUR_API_KEY>")model_config=Config(language=StreamLanguage())stream_options=StreamConnectOptions(config=model_config)asyncwithclient.expression_measurement.stream.connect(options=stream_options)assocket:forsampleinsamples:result=awaitsocket.send_text(sample)print(result.language.predictions[0]['emotions'])if__name__=="__main__":asyncio.run(main())
importasynciofromhumeimportHumeStreamClientfromhume.models.configimportLanguageConfigsamples= ["Mary had a little lamb,","Its fleece was white as snow.""Everywhere the child went,""The little lamb was sure to go." ]asyncdefmain():client=HumeStreamClient("<YOUR API KEY>")config=LanguageConfig()asyncwithclient.connect([config])assocket:forsampleinsamples:result=awaitsocket.send_text(sample)emotions=result["language"]["predictions"][0]["emotions"]print(emotions)if__name__=="__main__":asyncio.run(main())